Amazon Kinesis Data Firehose 数据转换 - Amazon Kinesis Data Firehose
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 Amazon Web Services 服务入门

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

Amazon Kinesis Data Firehose 数据转换

Kinesis Data Firehose 可以调用您的 Lambda 函数转换传入源数据并将转换后的数据传送到目标。当您创建传输流时,可以启用 Kinesis Data Firehose 数据转换。

数据转换流

启用 Kinesis Data Firehose 数据转换后,Kinesis Data Firehose 默认情况下将缓冲最多 3 MB 的传入数据。(要调整缓冲大小,请使用ProcessingConfiguration使用的 APIProcessorParameterCALLBufferSizeInMBs。) 然后,Kinesis Data Firehose 会将使用每个缓冲的批处理异步调用指定的 Lambda 函数Amazon Lambda同步调用模式。转换后的数据会从 Lambda 发送到 Kinesis Data Firehose。当达到指定的目标缓冲大小或缓冲间隔时(以先达到者为准),Kinesis Data Firehose 会将这些数据发送到目的地。

重要

Lambda 同步调用模式的请求和响应的负载大小限制为 6 MB。确保用于向函数发送请求的缓冲大小小于或等于 6 MB,并且函数返回的响应也不超过 6 MB。

数据转换和状态模型

所有通过 Lambda 转换的记录均包含以下参数,否则 Kinesis Data Firehose 会拒绝它们并将其视为数据转换失败。

recordId

记录 ID 在调用期间从 Kinesis Data Firehose 传递到 Lambda。转换后的记录必须包含相同记录 ID。原始记录的 ID 和转换记录的 ID 之间如果有不匹配,将被视为数据转换失败。

result

记录的数据转换的状态。可能的值为:Ok(记录成功转换)、Dropped(处理逻辑故意丢弃记录)和 ProcessingFailed(记录无法转换)。如果记录的状态为Ok或者Dropped时,Kinesis Data Firehose 认为它已成功处理。否则,Kinesis Data Firehose 会认为它未被成功处理。

data

转换后的数据负载 (使用 base64 编码之后)。

Lambda 蓝图

您可以使用蓝图为数据转换创建 Lambda 函数。其中一些蓝图位于 Amazon Lambda 控制台中,另一些位于中 Amazon Serverless Application Repository。

查看 Amazon Lambda 控制台中提供的蓝图

  1. 登录到 Amazon Web Services Management Console,然后通过以下网址打开 Amazon Lambda 控制台:https://console.aws.amazon.com/lambda/

  2. 选择 Create function (创建函数),然后选择 Use a blueprint (使用蓝图)

  3. 蓝图字段中,搜索关键字firehose查找 Kinesis Data Firehose Lambda 蓝图。

查看 Amazon Serverless Application Repository 中提供的蓝图

  1. 转到 Amazon Serverless Application Repository

  2. 选择浏览所有应用程序

  3. 应用程序 字段中,搜索关键字 firehose

您也可以在不使用蓝图的情况下创建 Lambda 函数。请参阅入门AmazonLambda

数据转换失败处理

如果您的 Lambda 函数调用由于网络超时或因为您已经达到 Lambda 调用限制而失败,Kinesis Data Firehose 会默认重试调用三次。如果调用不成功,Kinesis Data Firehose 随后会跳过该批记录。跳过的记录会被视为未被成功处理的记录。您可以使用 CreateDeliveryStreamUpdateDestination API 指定或覆盖重试选项。对于这种失败,您可以将调用错误记录到 Amazon CloudWatch Logs 中。有关更多信息,请参阅使用 CloudWatch Logs 监控 Kinesis Data Firehose

如果记录的数据转换的状态为ProcessingFailed,Kinesis Data Firehose 会将记录视为未被成功处理。对于这种失败,您可以从 Lambda 函数将错误日志发送到 Amazon CloudWatch Logs。有关更多信息,请参阅 。访问 Amazon CloudWatch LogsAmazon Lambda中的Amazon Lambda开发人员指南

如果数据转换失败,则未被成功处理的记录会传输到您在 processing-failed 文件夹中的 S3 存储桶。这些记录具有如下格式:

{ "attemptsMade": "count", "arrivalTimestamp": "timestamp", "errorCode": "code", "errorMessage": "message", "attemptEndingTimestamp": "timestamp", "rawData": "data", "lambdaArn": "arn" }
attemptsMade

尝试的调用请求数。

arrivalTimestamp

Kinesis Data Firehose 收到记录的时间。

errorCode

Lambda 返回的 HTTP 错误代码。

errorMessage

Lambda 返回的错误消息。

attemptEndingTimestamp

Kinesis Data Firehose 停止尝试 Lambda 调用的时间。

rawData

经过 base64 编码的记录数据。

lambdaArn

Lambda 函数的 Amazon 资源名称 (ARN)。

Lambda 调用的持续时间

Kinesis Data Firehose 支持的 Lambda 调用时间长达 5 分钟。如果您的 Lambda 函数需要超过 5 分钟才能完成,您将收到以下错误:消防管在调用时遇到超时错误AmazonLambda。支持的最大函数超时为 5 分钟。

有关在发生此类错误时 Kinesis Data Firehose 所执行的操作的信息,请参阅数据转换失败处理

源记录备份

Kinesis Data Firehose 可以在将转换后的记录传输到目标的同时,将所有未转换的记录备份到 S3 存储桶。您可以在创建或更新传输流时,启用源记录备份。在启用源记录备份之后,便不能禁用它。