亚马逊 Kinesis 数据 Firehose 构造库

AWS CDK v1在2023-06-01上达到了支持结束. 此软件包不再更新,用户应该迁移到 AWS CDK v2.

关于如何移徙的更多信息,请参见: 移动到 AWS CDK v2 指南。 。 。 .

亚马逊 Kinesis 数据火花 用于向存储服务全面提供实时流数据 例如Amazon S3, Amazon Redshift, Amazon Elsticsearch, Sprunk, 或任何自定义的 HTTP 等 终端或第三方服务,如Datadog、Dynatrace、LogicMonitor、MongoDB、New 遗物和苏莫逻辑.

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

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

定义传送流

要定义传送流, 您必须指定目的地 。 一个S3桶可以是 用作目的地。 涵盖更多受支助的目的地 下级。 。 。 .

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

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

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

来源

获取输入数据的主要方法有两种:Kinesis Data Streams和通过“Direct” 说吧.

见: 将数据发送到传送流 输入 Kinesis 数据发火者指南。 。 。 .

Kinesis 数据流

作为数据的消费者,可直接从Kinesis数据流读取数据流 流线。 通过在 sourceStream 构建传送流时的属性 :

declare const destination: firehose.IDestination;
const sourceStream = new kinesis.Stream(this, 'Source Stream');
new firehose.DeliveryStream(this, 'Delivery Stream', {
  sourceStream: sourceStream,
  destinations: [destination],
});

直接放入

数据必须通过“直接投放”即使用“直接投放”方式提供 PutRecord 或者说 PutRecordBatch API电话. 有几种办法 因此,例如:

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

目标

支持以下目的地. 见 火花点缀 执行这些目的地.

第3条

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

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

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

S3目的地也支持自定义动态前缀. prefix 将用于文件 成功交付到 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 前缀 输入 Kinesis 数据发火者指南。 。 。 .

服务器侧加密

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

用于SSE的密钥管理服务(KMS)客户管理密钥(CMK)可以是 AWS拥有或客户管理. AWS 拥有的 CMK 是 AWS 服务的密钥( 在这种情况下) Kinesis Data Firehose)拥有并管理用于多个AWS账户. 作为一个顾客, 您无法查看、使用、跟踪或管理这些密钥,您不会为这些密钥收取费用 使用时。 另一方面,客户管理的CMK是在内部创建和拥有的密钥 您的账户 完全由您管理。 作为客户,你负责管理 访问、旋转、别名和删除这些密钥,而您则因其 使用时。 见: 客户主密钥 输入 KMS 开发者指南。 。 。 .

declare const destination: firehose.IDestination;

// SSE with an AWS-owned CMK
new firehose.DeliveryStream(this, 'Delivery Stream AWS Owned', {
  encryption: firehose.StreamEncryption.AWS_OWNED,
  destinations: [destination],
});
// SSE with an customer-managed CMK that is created automatically by the CDK
new firehose.DeliveryStream(this, 'Delivery Stream Implicit Customer Managed', {
  encryption: firehose.StreamEncryption.CUSTOMER_MANAGED,
  destinations: [destination],
});
// SSE with an customer-managed CMK that is explicitly specified
declare const key: kms.Key;
new firehose.DeliveryStream(this, 'Delivery Stream Explicit Customer Managed', {
  encryptionKey: key,
  destinations: [destination],
});

见: 数据保护 内 页:1 Kinesis 数据发火者指南。 。 。 .

监测

Kinesis Data Firehose 与 Cloud Watch 集成, 这样您就可以监视 Cloud Watch 的性能 您通过日志和度量衡发送流 .

日志

Kinesis Data Firehose 数据转换或数据时会向 Cloud Watch 发送日志 交货失败。 CDK 默认启用日志并创建 Cloud Watch 日志组 和您的快递流的日志.

您可以提供一个特定的日志组, 指定 CDK 将在哪里创建日志流 发送日志事件:

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

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

declare const destination: firehose.IDestination;
new firehose.DeliveryStream(this, 'Delivery Stream', {
  destinations: [destination],
});

日志也可以禁用 :

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

见: 使用云表日志监测 输入 Kinesis 数据发火者指南。 。 。 .

计量

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

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

import * as cloudwatch from '@aws-cdk/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,
});

见: 利用云表量表进行监测 输入 Kinesis 数据发火者指南。 。 。 .

压缩

您的数据可自动被压缩, 当它作为最终或 中间/后备目的地。 支持的压缩格式有: 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', {
  destinations: [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', {
  destinations: [destination],
});

见: 数据传送频率 输入 Kinesis 数据发火者指南。 。 。 .

目标加密

您的数据在作为最终或 中间/后备目的地。 Kinesis 数据 Firehose 支持 Amazon S3 服务器侧 用 AWS 密钥管理服务加密已发送数据 在亚马逊S3。 您可以选择不加密数据或用密钥加密 您拥有的 AWS KMS 密钥列表。 更多信息见 保护数据 使用 AWS KMS - 管理的密钥( SSE- KMS) 加密服务器系统。数据默认不会加密.

declare const bucket: s3.Bucket;
declare const key: kms.Key;
const destination = new destinations.S3Bucket(bucket, {
  encryptionKey: key,
});
new firehose.DeliveryStream(this, 'Delivery Stream', {
  destinations: [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', {
  destinations: [
    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', {
  destinations: [
    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', {
  destinations: [
    new destinations.S3Bucket(bucket, {
      s3Backup: {
        mode: destinations.BackupMode.ALL,
        dataOutputPrefix: 'mybackup',
      },
    }),
  ],
});

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

数据处理/转换

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

与 AWS Lambda 数据转换

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

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

数据默认被缓冲至1分钟和3米B,然后发送到 函数,但可以用 bufferInterval 财务报告和审定财务报表 bufferSize 在处理器中 配置(见: 缓冲) (中文(简体) ). 如果由于下列原因未能援引函数: 或因达到援引限制,援引被重新审理 3 时间默认,但可以使用 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_14_X,
  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', {
  destinations: [s3Destination],
});

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

见: 数据转换 输入 Kinesis 数据发火者指南。 。 。 .

指定 IAM 角色

交付结构类自动创建所有最小的 IAM 服务角色 kinesis 数据 Firehose 访问您引用的资源的必要权限 送货流。 为允许 Kinesis 的送货流创建了一个服务角色 从 Kinesis 数据流读取的数据 Firehose (如果配置为发送) 用于服务器侧加密。 为每个单位设立了另一个服务角色 目的地,它允许 Kinesis Data Firehose 写入目的地资源, 以及调取数据变压器和读取记录格式的图案的能力 转换。 如果您愿意, 您可以指定您自己的 IAM 角色 或目的地服务角色,或两者兼有。 它必须有正确的信任政策(它必须: 允许 Kinesis 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 Kinesis 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', {
  destinations: [destination],
  role: deliveryStreamRole,
});

见 控制访问 输入 Kinesis 数据发火者指南。 。 。 .

允许应用程序访问传送流

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

执行 IGrantable 界面( e., 有相关主机) 可以通过调用:

• 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

授予资源传输流访问权限

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

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

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

多个目的地

虽然运送流允许指定一系列目的地,但只有一个 目前允许每个发送流的目的地。 此限制在 CDK 执行 合成时间, 并会投出错误 .