Lambda を使用したセルフマネージド Apache Kafka メッセージの処理

Kafka クラスターをイベントソースとして追加する

イベントソースマッピングを作成するには、Lambda コンソール、AWSSDK、またはAWS Command Line Interface(AWS CLI) を使用して、Kafka クラスターを Lambda 関数のトリガーとして追加します。

このセクションでは、Lambda コンソールと AWS CLI を使用してイベントソースマッピングを作成する方法について説明します。

前提条件

カスタマイズ可能なコンシューマーグループ ID

Kafkaをイベントソースとして設定する場合、コンシューマーグループIDを指定できます。このコンシューマーグループ ID は、Lambda 関数を結合したい Kafka コンシューマーグループの既存の識別子です。この機能を使用すると、実行中の Kafka レコード処理設定を他のコンシューマーから Lambda にシームレスに移行できます。

コンシューマーグループ ID を指定し、そのコンシューマーグループ内に他のアクティブなポーラーが存在する場合、Kafka はすべてのコンシューマーにメッセージを配信します。言い換えると、Lambda は Kafka トピックのメッセージをすべて受け取るわけではありません。Lambda にトピック内のすべてのメッセージを処理させたい場合は、そのコンシューマーグループの他のポーラーをすべてオフにします。

さらに、コンシューマーグループ ID を指定し、Kafka が同じ ID を持つ有効な既存のコンシューマーグループを見つけた場合、Lambda は、イベントソースマッピングの StartingPosition パラメーターを無視します。代わりに、Lambda はコンシューマーグループのコミットされたオフセットに従ってレコードの処理を開始します。コンシューマーグループ ID を指定しても、Kafka が既存のコンシューマーグループを見つけられない場合、Lambda は指定された StartingPosition を使用してイベントソースを設定します。

指定するコンシューマーグループ ID は、すべての Kafka イベントソースの中で一意でなければなりません。コンシューマーグループ ID を指定して Kafka イベントソースマッピングを作成した後は、この値を更新することはできません。

セルフマネージド型 Kafka クラスターを追加する (コンソール)

セルフマネージド型 Apache Kafka クラスターと Kafka トピックを Lambda 関数のトリガーとして追加するには、次の手順を実行します。

セルフマネージド型 Kafka クラスターを追加する (AWS CLI)

Lambda 関数のセルフマネージド型 Apache Kafka トリガーを作成および表示するには、次の AWS CLI コマンドの例を使用します。

SASL/SCRAM を使用する

Kafka ユーザーがインターネット経由で Kafka ブローカーにアクセスする場合は、SASL/SCRAM 認証用に作成した Secrets Manager シークレットを指定します。次の例では、create-event-source-mapping AWS CLI コマンドを使用して、my-kafka-function という名前の Lambda 関数を AWSKafkaTopic という名前の Kafka トピックにマッピングします。

aws lambda create-event-source-mapping \ 
  --topics AWSKafkaTopic \
  --source-access-configuration Type=SASL_SCRAM_512_AUTH,URI=arn:aws:secretsmanager:us-east-1:111122223333:secret:MyBrokerSecretName \
  --function-name arn:aws:lambda:us-east-1:111122223333:function:my-kafka-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["abc3.xyz.com:9092", "abc2.xyz.com:9092"]}}'

VPC の使用

Kafka ブローカーにアクセスするのが VPC 内の Kafka ユーザーのみである場合、VPC、サブネット、および VPC セキュリティグループを指定する必要があります。次の例では、create-event-source-mapping AWS CLI コマンドを使用して、my-kafka-function という名前の Lambda 関数を AWSKafkaTopic という名前の Kafka トピックにマッピングします。

aws lambda create-event-source-mapping \ 
  --topics AWSKafkaTopic \
  --source-access-configuration '[{"Type": "VPC_SUBNET", "URI": "subnet:subnet-0011001100"}, {"Type": "VPC_SUBNET", "URI": "subnet:subnet-0022002200"}, {"Type": "VPC_SECURITY_GROUP", "URI": "security_group:sg-0123456789"}]' \
  --function-name arn:aws:lambda:us-east-1:111122223333:function:my-kafka-function \
  --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["abc3.xyz.com:9092", "abc2.xyz.com:9092"]}}'

AWS CLI を使用したステータスの表示

次の例では、get-event-source-mapping AWS CLI コマンドを使用して、作成したイベントソースマッピングのステータスを記述します。

aws lambda get-event-source-mapping
          --uuid dh38738e-992b-343a-1077-3478934hjkfd7

セルフマネージド Apache Kafka の設定パラメータ

すべての Lambda イベントソースタイプは、同じ CreateEventSourceMapping および UpdateEventSourceMapping API オペレーションを共有しています。ただし、Apache Kafka に適用されるのは一部のパラメータのみです。

Kafka クラスターをイベントソースとして使用する

Apache Kafka クラスターまたは Amazon MSK クラスターを Lambda 関数のトリガーとして追加すると、クラスターはイベントソースとして使用されます。

Lambda は、ユーザーが指定した StartingPosition に基づいて、CreateEventSourceMapping リクエストで Topics として指定された Kafka トピックからイベントデータを読み取ります。処理が成功すると、Kafka トピックは Kafka クラスターにコミットされます。

StartingPosition を LATEST として指定すると、Lambda は、そのトピックに属する各パーティション内の最新のメッセージから読み取りを開始します。トリガーが設定されてから Lambda がメッセージの読み取りを開始するまでには若干の遅延が発生することがあるため、Lambda はこの期間中に生成されたメッセージを読み取りません。

Lambda は、指定された 1 つ、または複数の Kafka トピックパーティションからのレコードを処理し、関数に JSON ペイロードを送信します。1 つの Lambda ペイロードに、複数のパーティションからのメッセージを含めることができます。利用可能なレコードが増えると、Lambda は関数がトピックに追いつくまで、CreateEventSourceMapping リクエストで指定された BatchSize 値に基づいて、バッチ内のレコードの処理を継続します。

関数がバッチ内のいずれかのメッセージに対してエラーを返すと、Lambda は、処理が成功するかメッセージが期限切れになるまでメッセージのバッチ全体を再試行します。すべての再試行が失敗したレコードを、障害発生時の送信先に送信して、後で処理することができます。

ポーリングとストリームの開始位置

イベントソースマッピングの作成時および更新時のストリームのポーリングは、最終的に一貫性があることに注意してください。

つまり、LATEST をストリームの開始位置として指定すると、イベントソースマッピングの作成または更新中にイベントを見逃す可能性があります。イベントを見逃さないようにするには、ストリームの開始位置を TRIM_HORIZON または AT_TIMESTAMP として指定します。

セルフマネージド Apache Kafka イベントソースマッピングのメッセージスループットのスケーリング動作

Amazon MSK イベントソースマッピングでは、メッセージスループットのスケーリング動作を次の 2 つのモードから選択できます。

デフォルト (オンデマンド) モード

初めてセルフマネージド Apache Kafka イベントソースを作成すると、Lambda は Kafka トピック内のすべてのパーティションを処理するために、デフォルトの数のイベントポーラーを割り当てます。Lambda は、メッセージ負荷に基づいて、イベントポーラーの数を自動的にスケールアップまたはスケールダウンします。

Lambda は、1 分間隔でトピック内のすべてのパーティションのコンシューマーオフセット遅延を評価します。オフセットラグが大きすぎる場合、パーティションは Lambda で処理できる速度よりも速くメッセージを受信します。必要に応じて、Lambda はトピックのイベントポーラーを追加または削除します。イベントポーラーを追加または削除するこの自動スケーリングプロセスは、評価から 3 分以内に実行されます。

ターゲットの Lambda 関数がスロットリングされると、Lambda はイベントポーラーの数を減らします。このアクションにより、イベントポーラーが関数から取得するメッセージ、および関数に送信するメッセージの数が減ることで、関数に対する負荷が軽減されます。

Kafka トピックのスループットを監視するには、consumer_lag や consumer_offset などの Apache Kafka コンシューマーラグメトリックスを表示します。

プロビジョンドモードの設定

イベントソースマッピングのスループットを微調整する必要があるワークロードでは、プロビジョンドモードを使用できます。プロビジョンドモードでは、プロビジョニングされたイベントポーラーの最小数と最大数を定義します。これらのプロビジョニングされたイベントポーラーは、イベントソースマッピング専用であり、予期しないメッセージスパイクが発生した場合に、それらを瞬時に処理できます。パフォーマンス要件が厳しい Kafka ワークロードには、プロビジョンドモードを使用することをお勧めします。

Lambda のイベントポーラーは最大 5 MBps のスループットを処理できるコンピューティングユニットです。例えば、イベントソースの平均ペイロードが 1 MB で、平均関数所要時間が 1 秒であるとします。ペイロードで変換 (フィルタリングなど) が発生しない場合、1 つのポーラーが 5 MBps のスループットと 5 つの同時 Lambda 呼び出しをサポートできます。プロビジョニングモードを使用すると、追加コストが発生します。料金の見積もりについては、「AWS Lambda 料金」を参照してください。

プロビジョンドモードでは、イベントポーラーの最小数 (MinimumPollers) の許容値の範囲は 1～200 です。イベントポーラーの最大数 (MaximumPollers) の許容値の範囲は 1～2,000 です。MaximumPollers は MinimumPollers 以上である必要があります。また、パーティション内での順序付き処理を維持するために、Lambda は MaximumPollers をトピック内のパーティション数に制限します。

イベントポーラーの最小数および最大数に適切な値を選択する方の詳細については、「プロビジョンドモードを使用する際のベストプラクティスと考慮事項」を参照してください。

コンソールまたは Lambda API を使用して、セルフマネージド Apache Kafka イベントソースマッピングに対するプロビジョンドモードを設定できます。

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 '{}'

プロビジョンドモードを使用する際のベストプラクティスと考慮事項

イベントソースマッピングの最小および最大イベントポーラー数の最適な設定は、アプリケーションのパフォーマンス要件によって異なります。パフォーマンスプロファイルのベースラインを確立する際には、最初はデフォルトの最小イベントポーラー数から始めることをお勧めします。観測されたメッセージ処理パターンと望ましいパフォーマンスプロファイルに基づいて設定を調整します。

トラフィックの急増が発生しやすく、厳格なパフォーマンス要件があるワークロードの場合、メッセージの突然の急増を処理できるように、最小イベントポーラーを増やします。必要な最小イベントポーラー数を決定するには、ワークロードの 1 秒あたりのメッセージ数と平均ペイロードサイズを考慮し、1 つのイベントポーラーのスループットキャパシティ (最大 5 MBps) を参考にします。

パーティション内での順序付き処理を維持するために、Lambda は最大イベントポーラー数をトピック内のパーティション数に制限します。さらに、イベントソースマッピングがスケーリングできる最大イベントポーラー数は、関数の同時実行設定によって異なります。

プロビジョンドモードを有効にする際には、ネットワーク設定を更新して AWS PrivateLink VPC エンドポイントと関連するアクセス許可を削除します。

Amazon CloudWatch メトリクス

Lambda は、関数がレコードを処理している間に OffsetLag メトリクスを発行します。このメトリクスの値は、Kafka イベントソーストピックに書き込まれた最後のレコードと関数のコンシューマグループが処理した最後のレコードの間のオフセットの差分です。レコードが追加されてからコンシューマグループがそれを処理するまでのレイテンシーを見積もるには、OffsetLag を使用できます。

OffsetLag の増加傾向は、関数のコンシューマグループに問題があることを示している可能性があります。詳細については、「Lambda での CloudWatch メトリクスの使用」を参照してください。