使用 Lambda 处理 Amazon MSK 消息

将 Amazon MSK 添加为事件源

创建事件源映射，使用 Lambda 控制台、Amazon开发工具包，或 Amazon Command Line Interface (Amazon CLI) 将您的 Amazon MSK 添加为 Lambda 函数触发器。请注意，当您将 Amazon MSK 添加为触发器时，Lambda 将假定 Amazon MSK 集群的 VPC 设置，而不是 Lambda 函数的 VPC 设置。

本节介绍了如何使用 Lambda 控制台和 Amazon CLI 创建事件源映射。

先决条件

可自定义的使用者组 ID

将 Kafka 设置为事件源时，您可以指定使用者组 ID。此使用者组 ID 是您希望 Lambda 函数加入的 Kafka 使用者组的现有标识符。您可以使用此功能将任何正在进行的 Kafka 记录处理设置从其他使用者无缝迁移到 Lambda。

如果指定了使用者组 ID，并且该使用者组中还有其他活跃的轮询器，则 Kafka 会向所有使用者分发消息。换句话说，Lambda 不会收到 Kafka 主题的所有消息。如果希望 Lambda 处理主题中的所有消息，请关闭该使用者组中的任何其他轮询器。

此外，如果指定了使用者组 ID，而 Kafka 找到了具有相同 ID 的有效现有使用者组，则 Lambda 会忽略事件源映射的 StartingPosition 参数。相反，Lambda 开始根据使用者组的已提交偏移量处理记录。如果指定了使用者组 ID，而 Kafka 找不到现有使用者组，则 Lambda 会使用指定的 StartingPosition 配置事件源。

在所有 Kafka 事件源中，您指定的使用者组 ID 必须是唯一的。在使用指定的使用者组 ID 创建 Kafka 事件源映射后，无法更新此值。

添加 Amazon MSK 触发器（控制台）

按照以下步骤将 Amazon MSK 集群和 Kafka 主题添加为 Lambda 函数的触发器。

添加 Amazon MSK 触发器（Amazon CLI）

使用以下示例 Amazon CLI 命令为 Lambda 函数创建和查看 Amazon MSK 触发器。

使用 Amazon CLI 创建触发器

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function
  --source-access-configurations '[{"Type": "SASL_SCRAM_512_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function
  --source-access-configurations '[{"Type": "CLIENT_CERTIFICATE_TLS_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'

有关更多信息，请参阅 CreateEventSourceMapping API 参考文档。

使用 Amazon CLI 查看状态

以下示例使用 get-event-source-mapping Amazon CLI 命令来描述您创建的事件源映射的状态。

aws lambda get-event-source-mapping \
  --uuid 6d9bce8e-836b-442c-8070-74e77903c815

Amazon MSK 配置参数

所有 Lambda 事件源类型共享相同的 CreateEventSourceMapping 和 UpdateEventSourceMapping API 操作。但是，只有部分参数适用于Amazon MSK。

创建跨账户事件源映射

您可以使用多 VPC 私有连接将 Lambda 函数连接到不同 Amazon Web Services 账户 中的预置 MSK 集群。多 VPC 连接使用 Amazon PrivateLink，可将所有流量保持在 Amazon 网络内。

要创建跨账户事件源映射，必须先为 MSK 集群配置多 VPC 连接。创建事件源映射时，请使用托管 VPC 连接 ARN 而非集群 ARN，如以下示例所示。CreateEventSourceMapping 操作也因 MSK 集群使用的身份验证类型而异。

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function \
  --source-access-configurations '[{"Type": "SASL_SCRAM_512_AUTH","URI": "arn:aws:secretsmanager:us-east-1:444455556666:secret:my-secret"}]'

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function \
  --source-access-configurations '[{"Type": "CLIENT_CERTIFICATE_TLS_AUTH","URI": "arn:aws:secretsmanager:us-east-1:444455556666:secret:my-secret"}]'

将 Amazon MSK 集群用作事件源

当您添加 Apache Kafka 或 Amazon MSK 集群作为 Lambda 函数的触发器时，该集群将用作事件源。

Lambda 根据您指定的 StartingPosition，从您在 CreateEventSourceMapping 请求中指定为 Topics 的 Kafka 主题读取事件数据。成功进行处理后，会将 Kafka 主题提交给 Kafka 集群。

如果您指定 StartingPosition 作为 LATEST，则 Lambda 开始读取主题下每个分区中的最新消息。由于在触发器配置后 Lambda 开始读取消息之前可能会有一些延迟，因此 Lambda 不会读取在此窗口中生成的任何消息。

Lambda 按顺序读取每个 Kafka 主题分区的消息。单个 Lambda 负载可以包含来自多个分区的消息。当有更多记录可用时，Lambda 根据您在 CreateEventSourceMapping 中指定的 BatchSize 值，继续对记录进行批处理，直到函数赶上主题的速度。

Lambda 处理各个批次后，会提交该批次中消息的偏移量。如果函数为批处理中的任何消息返回错误，Lambda 将重试整批消息，直到处理成功或消息过期为止。您可以将所有重试都失败的记录发送到失败时的目标，以供日后处理。

轮询和流的起始位置

请注意，事件源映射创建和更新期间的流轮询最终是一致的。

此行为意味着，如果你指定 LATEST 作为流的起始位置，事件源映射可能会在创建或更新期间错过事件。为确保不会错过任何事件，请将流的起始位置指定为 TRIM_HORIZON 或 AT_TIMESTAMP。

Amazon CloudWatch 指标

Lambda 会在您的函数处理记录时发出 OffsetLag 指标。此指标的值是写入 Kafka 事件源主题的最后一条记录与函数的使用者组处理的最后一条记录之间的偏移量差值。您可以使用 OffsetLag 来估计添加记录和使用者组处理记录之间的延迟。

如果 OffsetLag 呈上升趋势，则可能表明函数的使用者组中的轮询器存在问题。有关更多信息，请参阅 将 CloudWatch 指标与 Lambda 结合使用。

Amazon MSK 事件源映射的消息吞吐量扩展行为

您可以在 Amazon MSK 事件源映射的两种消息吞吐量扩展行为模式之间进行选择：

默认（按需）模式

当您最初创建 Amazon MSK 事件源时，Lambda 会分配默认数量的事件轮询器来处理 Kafka 主题中的所有分区。Lambda 根据消息负载自动扩展或缩减事件轮询器的数量。

Lambda 会按一分钟的间隔时间来评估主题中所有分区的偏移滞后。如果偏移延迟太高，则分区接收消息的速度比 Lambda 处理消息的速度更快。如有必要，Lambda 会在主题中添加或删除事件轮询器。添加或删除事件轮询器的自动扩缩过程在评估后的三分钟内发生。

如果目标 Lambda 函数受到限制，Lambda 会减少事件轮询器的数量。此操作通过减少事件轮询器可以检索和发送到函数的消息数来减少函数的工作负载。

配置预调配模式

对于需要微调事件源映射吞吐量的工作负载，您可以使用预调配模式。在预调配模式下，您可以为预调配事件轮询器数量定义最小和最大限制。这些预调配事件轮询器专用于事件源映射，并且可以通过响应式自动扩缩处理意外的消息激增。对于具有严格性能要求的 Kafka 工作负载，我们建议您使用预调配模式。

在 Lambda 中，事件轮询器是一种能够处理高达 5 Mbps 吞吐量的计算单位。作为参考，假设您的事件源产生的平均有效载荷为 1 MB，并且平均函数持续时间为 1 秒。如果有效载荷未进行任何转换（例如筛选），则单个轮询器可以支持 5 Mbps 的吞吐量和 5 次并发 Lambda 调用。使用预调配模式会产生额外成本。有关定价估算，请参阅 Amazon Lambda 定价。

在预调配模式下，最小事件轮询器数量 (MinimumPollers) 的可接受值范围介于 1 到 200 之间（含首尾）。事件轮询器的最大数量 (MaximumPollers) 的可接受值范围介于 1 到 2000 之间（含首尾）。MaximumPollers 必须大于或等于 MinimumPollers。此外，为了保持分区内的有序处理，Lambda 會将 MaximumPollers 限制为主题中的分区数量。

有关选择适当的最小和最大事件轮询器值的更多详细信息，请参阅使用预调配模式时的最佳实践和注意事项。

您可以使用控制台或 Lambda API 为 Amazon MSK 事件源映射配置预调配模式。

您可以使用 EventSourceMappingConfiguration 中的 ProvisionedPollerConfig 对象，以编程方式配置预调配模式。例如，以下 UpdateEventSourceMapping CLI 命令将 MinimumPollers 值配置为 5，将 MaximumPollers 值配置为 100。

aws lambda update-event-source-mapping \
    --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \
    --provisioned-poller-config '{"MinimumPollers": 5, "MaximumPollers": 100}'

配置预调配模式后，您可以通过监控 ProvisionedPollers 指标来观测事件轮询器对您的工作负载的使用情况。有关更多信息，请参阅 事件源映射指标。

要禁用预调配模式并返回默认（按需）模式，您可以使用以下 UpdateEventSourceMapping CLI 命令：

aws lambda update-event-source-mapping \
    --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \
    --provisioned-poller-config '{}'

使用预调配模式时的最佳实践和注意事项

事件源映射的最小和最大事件轮询器的最佳配置取决于应用程序的性能需求。建议您从默认最小事件轮询器开始，以设定性能配置文件的基准。根据观测到的消息处理模式和所需的性能配置文件调整配置。

对于流量激增且性能需求严格的工作负载，请增加最少的事件轮询器数以处理消息突然激增。要确定所需的最少事件轮询器数，请考虑工作负载的每秒消息数和平均有效载荷大小，并使用单个事件轮询器的吞吐能力（最高 5 Mbps）作为参考。

为了保持分区内的有序处理，Lambda 会将最大事件轮询器数限制为主题中的分区数量。此外，您的事件源映射可以扩展到的最大事件轮询器数取决于函数的并发设置。

激活预调配模式时，更新您的网络设置以删除 Amazon PrivateLink VPC 端点和关联的权限。