亚马逊数据 Firehose 构造库

此API 可能会发出警告 。 后向兼容性没有保证 .

所有构造都转移到aws-cdk-lib/aws-kinesisfirehose.

亚马逊数据Firehose, (中文(简体) ), 原名"亚马逊克尼斯斯数据"Firehose, (中文(简体) ), 用于向存储服务全面提供实时流数据 例如Amazon S3, Amazon Redshift, Amazon Elsticsearch, Sprunk, 或任何自定义的 HTTP 等 终端或第三方服务,如Datadog、Dynatrace、LogicMonitor、MongoDB、New 遗物和苏莫逻辑.

Amazon Data Firehose 传送流与 Kinesis 数据流在其中区分 他们的消费模式。 消费者通过积极拉动从数据流读取 来自溪流的数据,一个送出溪流将数据推向其目的地的常规 ca. 这意味着数据流旨在让消费者进行点播 加工,如AWS Lambda或亚马逊EC2. 另一方面,送货流是 意图使目的地成为离线处理和分析的来源 作为亚马逊S3和亚马逊Redshift.

这个模块是 AWS 云开发套件 项目。 它允许您定义 Amazon 数据 Firehose 发送流 .

定义传送流

要定义传送流, 您必须指定目的地 。 一个S3桶可以是 用作目的地。 目前, CDK 仅支持 S3 作为覆盖的目的地 下级。 。 。 .

const bucket = new s3.Bucket(this, 'Bucket');
new firehose.DeliveryStream(this, 'Delivery Stream', {
  destination: new destinations.S3Bucket(bucket),
});

以上例子界定了下列资源:

• 一个S3桶
• 以直接 PUT 为源代码和 Cloud Watch 的 Amazon 数据 Firehose 传送流 打开了错误记录 .
• 一个IAM角色,它允许送货流写给S3桶.

来源

一个亚马逊数据Firehose发送流可以接受来自三个主要来源的数据:Kinesis Data Streams,为Apache Kafka(MSK)管理Streaming,或者通过"直接放送"(API调用). 目前只支持Kinesis数据流和直接投放.

见: 将数据发送到传送流 输入 亚马逊数据 Firehose 开发者指南。 。 。 .

Kinesis 数据流

作为数据的消费者,可直接从Kinesis数据流读取数据流 流线。 通过在其中的数据流中传递来配置此行为 source 通过 KinesisStreamSource 构造传送流时的分类 :

declare const destination: firehose.IDestination;
const sourceStream = new kinesis.Stream(this, 'Source Stream');

new firehose.DeliveryStream(this, 'Delivery Stream', {
  source: new firehose.KinesisStreamSource(sourceStream),
  destination: destination,
});

直接放入

数据必须通过“直接投放”即使用“直接投放”方式提供 PutRecord 或者说 PutRecordBatch API电话. 这样做有许多方法,例如:

• Kinesis Agent: 一个独立 Java 应用程序,用于监视和发送文件同时 处理文件旋转、检查和重试。 见: 使用 Kinesis 代理写入 Amazon 数据 Firehose 输入 亚马逊数据 Firehose 开发者指南。 。 。 .
• AWS SDK:一个允许您将数据传送到传输流的通用解决方案 从任何地方使用Java,.NET,节点js,Python,或Ruby. 见: 使用 AWS SDK 写入 Amazon 数据 Firehose 输入 亚马逊数据 Firehose 开发者指南。 。 。 .
• Cloud Watch 日志: 订阅日志组并直接接收过滤到的日志事件 一个快递流。 见: 日志目的地。 。 。 .
• 事件桥: 添加事件规则目标, 将事件发送到基于 规则过滤。 见: 事件目标。 。 。 .
• SNS: 添加订阅, 将专题中的所有通知发送到发送 流线。 见: sns 订阅。 。 。 .
• IOT:在IOT规则中添加一个动作,将各种IOT信息发送到快递流中

目标

Amazon Data Firehose支持多个AWS和第三方服务作为目的地,包括Amazon S3,Amazon Redshift等. 您可以找到支持目的的全部列表 这里。 。 。 .

目前,在AWS CDK中,只有S3作为L2构造目的地被执行. 其他目的地仍然可以使用L1构造进行配置. 见 火花点缀 执行这些目的地.

第3条

使用 S3 桶目标定义送货流 :

declare const bucket: s3.Bucket;
const s3Destination = new destinations.S3Bucket(bucket);

new firehose.DeliveryStream(this, 'Delivery Stream', {
  destination: s3Destination,
});

S3目的地也支持自定义动态前缀. dataOutputPrefix 将被用于成功交付到 S3 的文件 . errorOutputPrefix 将添加到 失败的记录在写入 S3 之前 .

declare const bucket: s3.Bucket;
const s3Destination = new destinations.S3Bucket(bucket, {
  dataOutputPrefix: 'myFirehose/DeliveredYear=!{timestamp:yyyy}/anyMonth/rand=!{firehose:random-string}',
  errorOutputPrefix: 'myFirehoseFailures/!{firehose:error-output-type}/!{timestamp:yyyy}/anyMonth/!{timestamp:dd}',
});

见: 自定义 S3 前缀 输入 亚马逊数据 Firehose 开发者指南。 。 。 .

服务器侧加密

启用服务器侧加密( SSE) 需要 Amazon Data Firehose 加密所有数据 当它储存在休息时, 发送到发送流。 这意味着数据是加密的 在被写入服务内部存储层之前, 然后解密 从内部存储层收到。 服务管理密钥和密码 操作,使源和目的地无需,因为数据是加密的 在服务边界被解密(即在数据交付给一个 目标。 默认情况下, 交付流没有启用 SSE .

SSE 使用的密钥管理服务密钥( KMS key) 可以是 AWS 拥有的或 客户管理。 AWS 拥有的 KMS 密钥由 AWS 创建、拥有和管理, 用于 多个AWS账户. 作为客户,你不能查看,使用,跟踪,或管理这些密钥, 而你们不因他们的使用而受罚。 另一方面,客户管理的 KMS 密钥是 在你的账户里创建和拥有 完全由你管理 作为顾客,你是 负责管理这些密钥的访问、旋转、别名和删除,以及您 用于修改.

见: AWS KMS 密钥 输入 KMS 开发者指南。 。 。 .

declare const destination: firehose.IDestination;

// SSE with an AWS-owned key
new firehose.DeliveryStream(this, 'Delivery Stream with AWS Owned Key', {
  encryption: firehose.StreamEncryption.awsOwnedKey(),
  destination: destination,
});
// SSE with an customer-managed key that is created automatically by the CDK
new firehose.DeliveryStream(this, 'Delivery Stream with Customer Managed Key', {
  encryption: firehose.StreamEncryption.customerManagedKey(),
  destination: destination,
});
// SSE with an customer-managed key that is explicitly specified
declare const key: kms.Key;
new firehose.DeliveryStream(this, 'Delivery Stream with Customer Managed and Provided Key', {
  encryption: firehose.StreamEncryption.customerManagedKey(key),
  destination: destination,
});

见: 数据保护 输入 亚马逊数据 Firehose 开发者指南。 。 。 .

监测

Amazon Data Firehose 与 Cloud Watch 集成后,您可以监控其性能 您通过日志和度量衡发送流 .

日志

Amazon Data Firehose 数据转换或数据时会向 Cloud Watch 发送日志 交货失败。 CDK 默认启用日志并创建 Cloud Watch 日志组 和带有默认设置的日志流.

当创建目的地时,您可以提供 ILoggingConfig,可以是 EnableLogging 或者说 DisableLogging 举个例子. 如果您使用 EnableLogging, CDK 将会为您创建包含所有 CloudFormation 默认设置的 Cloud Watch 日志组和日志结构, 或者您可以选择 指定用于记录和存储日志事件的自身日志组。 例如:

import * as logs from 'aws-cdk-lib/aws-logs';

const logGroup = new logs.LogGroup(this, 'Log Group');
declare const bucket: s3.Bucket;
const destination = new destinations.S3Bucket(bucket, {
  loggingConfig: new destinations.EnableLogging(logGroup),
});

new firehose.DeliveryStream(this, 'Delivery Stream', {
  destination: destination,
});

日志也可以禁用 :

declare const bucket: s3.Bucket;
const destination = new destinations.S3Bucket(bucket, {
  loggingConfig: new destinations.DisableLogging(),
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
  destination: destination,
});

见: 使用云表日志监测 输入 亚马逊数据 Firehose 开发者指南。 。 。 .

计量

Amazon Data Firehose 向 Cloud Watch 发送度量衡,以便收集和分析 包括数据发送、数据摄入、数据 转换、格式转换、API使用、加密和资源使用。 那你们可以了 使用 Cloud Watch 提醒提醒您, 例如当数据新鲜度( 时间为 执行流中最古老的记录)超过缓冲限制(表明数据是 如果收到记录的比率超过 每秒记录的限度(指示数据会更快地流入您的交付流) 而非它被配置为处理).

CDK 提供了使用默认配置访问交付流度量的方法, 例如 metricIncomingBytes,以及 metricIncomingRecords (见 IDeliveryStream 为完整的名单。 CDK 也提供了一个通用 metric 可用于生产的方法 amazon Data Firehose提供的任何度量的度量配置; 配置 以正确的尺寸预选送货流.

import * as cloudwatch from 'aws-cdk-lib/aws-cloudwatch';

declare const deliveryStream: firehose.DeliveryStream;

// Alarm that triggers when the per-second average of incoming bytes exceeds 90% of the current service limit
const incomingBytesPercentOfLimit = new cloudwatch.MathExpression({
  expression: 'incomingBytes / 300 / bytePerSecLimit',
  usingMetrics: {
    incomingBytes: deliveryStream.metricIncomingBytes({ statistic: cloudwatch.Statistic.SUM }),
    bytePerSecLimit: deliveryStream.metric('BytesPerSecondLimit'),
  },
});

new cloudwatch.Alarm(this, 'Alarm', {
  metric: incomingBytesPercentOfLimit,
  threshold: 0.9,
  evaluationPeriods: 3,
});

见: 利用云表量表进行监测 输入 亚马逊数据 Firehose 开发者指南。 。 。 .

压缩

您的数据可自动被压缩, 当它作为最终或 中间/后备目的地。 支持的压缩格式有: gzip, Snappy, 相容的相容 Snappy 和 ZIP, 除了 Redshift 目的地, Snappy (无论Hadoop兼容性如何)和ZIP不支持. 默认数据为 发送到 S3 不压缩.

// Compress data delivered to S3 using Snappy
declare const bucket: s3.Bucket;
const s3Destination = new destinations.S3Bucket(bucket, {
  compression: destinations.Compression.SNAPPY,
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
  destination: s3Destination,
});

缓冲

接收数据在交付到指定目的地之前进行缓冲. 这个 发送流将等待数据数量超过某个阈值 (“缓冲大小”)或直到上次数据发送发生时间超过 阈值("缓冲间隔"),以先发生者为准。 您可以配置这些 根据目的地的能力和您的使用量设定阈值 。 默认情况下 缓冲大小为5 MiB,缓冲间隔为5分钟.

// Increase the buffer interval and size to 10 minutes and 8 MiB, respectively
declare const bucket: s3.Bucket;
const destination = new destinations.S3Bucket(bucket, {
  bufferingInterval: Duration.minutes(10),
  bufferingSize: Size.mebibytes(8),
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
  destination: destination,
});

见: 数据传送频率 输入 亚马逊数据 Firehose 开发者指南。 。 。 .

零缓冲,亚马逊数据 Firehose 流在发送前可以配置为不缓冲数据,由 将“缓冲间隔”设置为 0.

// Setup zero buffering
declare const bucket: s3.Bucket;
const destination = new destinations.S3Bucket(bucket, {
  bufferingInterval: Duration.seconds(0),
});
new firehose.DeliveryStream(this, 'ZeroBufferDeliveryStream', {
  destination: destination,
});

见: 缓冲提示。 。 。 .

目标加密

您的数据可自动加密, 当它作为最终或最终交付给 S3 时 中间/备份目的地。 Amazon Data Firehose 支持 Amazon S3 服务器侧 使用 AWS 密钥管理服务加密已发送的数据 亚马逊S3. 您可以选择不加密数据或用列表中的密钥加密 您拥有的 AWS KMS 密钥 。 有关更多信息, 见 使用 AWS KMS- 管理的密钥( SSE- KMS) 保护数据。 。 。 . 默认情况下, 加密不会直接在发送流上被启用; 而是使用目的地 S3 桶的默认加密设置 .

declare const bucket: s3.Bucket;
declare const key: kms.Key;
const destination = new destinations.S3Bucket(bucket, {
  encryptionKey: key,
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
  destination: destination,
});

备份

可以配置一个传送流,以备份它试图传送到的S3的数据 所配置的目的地。 备份数据可以成为发送流的全部数据 试图提供或仅仅提供它未能提供的数据(红调和S3目的地) 只能备份所有数据). CDK 可以创建一个新的 S3 桶, 用于备份数据, 或者 您可以提供数据备份的桶。 您也可以提供前缀 将您的备份数据放入桶中。 默认情况下, 源数据不是 备份到S3.

// Enable backup of all source records (to an S3 bucket created by CDK).
declare const bucket: s3.Bucket;
new firehose.DeliveryStream(this, 'Delivery Stream Backup All', {
  destination:
    new destinations.S3Bucket(bucket, {
      s3Backup: {
        mode: destinations.BackupMode.ALL,
      },
    }),
});
// Explicitly provide an S3 bucket to which all source records will be backed up.
declare const backupBucket: s3.Bucket;
new firehose.DeliveryStream(this, 'Delivery Stream Backup All Explicit Bucket', {
  destination: 
    new destinations.S3Bucket(bucket, {
      s3Backup: {
        bucket: backupBucket,
      },
    }),
});
// Explicitly provide an S3 prefix under which all source records will be backed up.
new firehose.DeliveryStream(this, 'Delivery Stream Backup All Explicit Prefix', {
  destination:
    new destinations.S3Bucket(bucket, {
      s3Backup: {
        mode: destinations.BackupMode.ALL,
        dataOutputPrefix: 'mybackup',
      },
    }),
});

如果在您的传送流上配置了任何数据处理或转换,则来源 记录将以原始格式备份.

数据处理/转换

数据可以在交付目的地之前被转换. 有两种类型 用于传输流的数据处理:与 AWS Lambda 进行记录转换,并记录 使用存储在 AWS Glue 表格中的 schema 格式转换。 如果这两种数据都是 处理是配置的,然后首先进行Lambda转换。 默认, 没有进行数据处理。 此构建库目前只支持数据 与AWS Lambda相变. 见 #15501 (英语) 以跟踪为记录格式转换添加支持的状态.

与 AWS Lambda 数据转换

为了转换数据,Amazon Data Firehose 将调用您提供的 Lambda 函数 并提供返回的数据取而代之的源记录。 该函数必须返回 a 包含特定格式的记录,包括下列字段:

• recordId -- 对应结果的输入记录的ID.
• result -?),记录的变换状态:"Ok"(成功),"被" (未故意处理),或"处理失败"(因出错未处理).
• data - 转换的数据, Base64 - 编码.

数据默认被缓冲至1分钟和3米B,然后发送到 函数,但可以用 bufferInterval 财务报告和审定财务报表 bufferSize 在处理器配置中(参见: 缓冲) (中文(简体) ). 如果函数引用 因网络超时或因达到援引限制而失败 默认重试三次,但可以使用 retries 在处理器中 配置 .

// Provide a Lambda function that will transform records before delivery, with custom
// buffering and retry configuration
const lambdaFunction = new lambda.Function(this, 'Processor', {
  runtime: lambda.Runtime.NODEJS_LATEST,
  handler: 'index.handler',
  code: lambda.Code.fromAsset(path.join(__dirname, 'process-records')),
});
const lambdaProcessor = new firehose.LambdaFunctionProcessor(lambdaFunction, {
  bufferInterval: Duration.minutes(5),
  bufferSize: Size.mebibytes(5),
  retries: 5,
});
declare const bucket: s3.Bucket;
const s3Destination = new destinations.S3Bucket(bucket, {
  processor: lambdaProcessor,
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
  destination: s3Destination,
});

实例 Lambda 数据处理器进行身份转换.

见: 数据转换 输入 亚马逊数据 Firehose 开发者指南。 。 。 .

指定 IAM 角色

交付结构类自动创建所有最小的 IAM 服务角色 亚马逊数据 Firehose 访问您引用的资源的必要权限 送货流。 为允许亚马逊的传送流创建了一个服务角色 从 Kinesis 数据流读取的数据 Firehose (如果配置为发送) 用于服务器侧加密。 注意, 如果创建了“ 交付结构 ” 不指定 source 或者说 encryptionKey,这个角色并非因为不需要而产生.

为每个目的地创建了另一个服务角色, 这使得 Amazon Data Firehose 写入 使用目的地资源,以及使用数据变压器和 读取用于记录格式转换的策略。 如果你愿意,你可以指定自己的IAM角色,用于 或交付流或目的地服务角色,或两者兼有。 一定没错 信任政策(必须允许Amazon Data Firehose承担)或交付流创建或 数据传输会失败。 目的地资源的其他必需权限,加密密钥等, 将自动提供.

// Create service roles for the delivery stream and destination.
// These can be used for other purposes and granted access to different resources.
// They must include the Amazon Data Firehose service principal in their trust policies.
// Two separate roles are shown below, but the same role can be used for both purposes.
const deliveryStreamRole = new iam.Role(this, 'Delivery Stream Role', {
  assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
});
const destinationRole = new iam.Role(this, 'Destination Role', {
  assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
});

// Specify the roles created above when defining the destination and delivery stream.
declare const bucket: s3.Bucket;
const destination = new destinations.S3Bucket(bucket, { role: destinationRole });
new firehose.DeliveryStream(this, 'Delivery Stream', {
  destination: destination,
  role: deliveryStreamRole,
});

见 控制访问 输入 亚马逊数据 Firehose 开发者指南。 。 。 .

允许应用程序访问传送流

IAM的作用、用户或团体需要能够与交付流程合作的,应当是 获得IAM许可.

执行 IGrantable 接口(即有相关主机) 可以通过调用:

• grantPutRecords(principal) - 使校长能够将记录输入 发送流
• grant(principal, ...actions) - 授予一套自定义的主要许可: 动作

const lambdaRole = new iam.Role(this, 'Role', {
  assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
});

// Give the role permissions to write data to the delivery stream
declare const deliveryStream: firehose.DeliveryStream;
deliveryStream.grantPutRecords(lambdaRole);

下列写作权限由服务主管提供: grantPutRecords() 方法 :

• firehose:PutRecord
• firehose:PutRecordBatch

授予资源传输流访问权限

反之,亚马逊数据Firehose 需要权限才能交付 流与您拥有的资源交互。 例如,如果指定 S3 桶 作为送货流的目的地,必须准许送货流 放在桶里取东西 使用内置 AWS 服务目的地时 发现于 @aws-cdk/aws-kinesisfirehose-destinations-alpha 模块中, CDK 授予 自动权限。 但是,习惯目的地或第三方目的地可能要求习惯目的地 权限。 这样的话,就把送货流当作 IGrantable,内容如下:

const fn = new lambda.Function(this, 'Function', {
  code: lambda.Code.fromInline('exports.handler = (event) => {}'),
  runtime: lambda.Runtime.NODEJS_LATEST,
  handler: 'index.handler',
});

declare const deliveryStream: firehose.DeliveryStream;
fn.grantInvoke(deliveryStream);