出力としての Lambda 関数の使用

AWS Lambda を宛先として使用すると、最終的な宛先への送信に先立つ SQL 結果の後処理がより簡単になります。一般的な後処理タスクには次のものがあります。

• 複数の行を 1 つのレコードに集約する

• 現在の結果と過去の結果を組み合わせて、遅れて届くデータに対処する

• 情報のタイプに基づいて異なる送信先に配信する

• レコード形式の変換 (Protobuf への変換など)

• 文字列操作または変換

• 分析処理後のデータの強化

• 地理空間ユースケースのカスタム処理

• データ暗号化

Lambda 関数は、AWS次のようなさまざまなサービスやその他の宛先に分析情報を配信できます。

• Amazon Simple Storage Service (Amazon S3)

• カスタム API

• Amazon DynamoDB

• Apache Aurora

• Amazon Redshift

• Amazon Simple Notification Service (Amazon SNS)

• Amazon Simple Queue Service (Amazon SQS)

• アマゾン CloudWatch

Lambda アプリケーション作成の詳細については、「使用開始」を参照してくださいAWS Lambda。

トピック

• 出力許可としての Lambda
• 出力メトリクスとしてのLambda
• 出力イベント入力データモデルおよびレコードレスポンスモデルとしての Lambda
• Lambda 出力呼び出しの頻度
• 出力として使用する Lambda 関数の追加
• よくあるLambda AS出力の失敗
• アプリケーション送信先の Lambda 関数を作成

出力許可としての Lambda

Lambda を出力として使用するには、アプリケーションの Lambda 出力 IAM ロールに次の権限ポリシーが必要です。

{
   "Sid": "UseLambdaFunction",
   "Effect": "Allow",
   "Action": [
       "lambda:InvokeFunction",
       "lambda:GetFunctionConfiguration"
   ],
   "Resource": "FunctionARN"
}

出力メトリクスとしてのLambda

Amazon CloudWatch を使用して、送信されたバイト数、成功と失敗などを監視します。Lambda を出力として使用して Kinesis Data Analytics CloudWatch によって生成されるメトリックスについては、「Amazon Kinesis アナリティクスのメトリクス」を参照してください。

出力イベント入力データモデルおよびレコードレスポンスモデルとしての Lambda

Kinesis Data Analytics の出力レコードを送信するには、Lambda 関数が必要なイベント入力データとレコード応答モデルに準拠している必要があります。

イベント入力データモデル

Kinesis Data Analytics は、次のリクエストモデルを使用して、アプリケーションからの出力レコードを出力関数として Lambda に継続的に送信します。関数内では、リストを繰り返し処理し、ビジネスロジックを適用して、出力要件 (最終的な送信先に送信する前のデータ変換など) を実行します。

フィールド	説明
invocationId	Lambda 呼び出しID (ランダム GUID)。
applicationArn	Kinesis Data Analytics アプリケーションの Amazon リソースネーム (ARN)。
レコード フィールド 説明 recordId レコード ID (ランダム GUID) lambdaDeliveryRecordMetadata フィールド 説明 retryHint 配信再試行回数 データ Base64 でエンコードされた出力レコードのペイロード

注記

retryHint は配信失敗ごとに増加する値です。この値は永続的に保持されず、アプリケーションが中断された場合にリセットされます。

レコードレスポンスモデル

出力関数 (レコード ID を含む) として Lambda に送信される各レコードには、Okまたはのいずれかで確認する必要がありDeliveryFailed、次のパラメータが含まれている必要があります。そうしないと、Kinesis Data Analytics それらを配信失敗として扱います。

レコード フィールド 説明 recordId レコード ID は、呼び出し時に Kinesis Data Analytics から Lambda に渡されます。元のレコードの ID と確認されたレコードの ID との不一致は、配信失敗として扱われます。 result レコード配信のステータス。以下の値を指定できます。 Ok: レコードは正常に変換され、最終宛先に送信されました。Kinesis Data Analytics は SQL 処理のレコードを取り込みます。 DeliveryFailed: レコードは、Lambda as output 関数によって最終宛先に正常に配信されませんでした。Kinesis Data Analytics は、配信に失敗したレコードを出力関数として Lambda に送信することを継続的に再試行します。

Lambda 出力呼び出しの頻度

Kinesis Data Analytics アプリケーションは出力レコードをバッファリングし、AWS Lambda宛先関数を頻繁に呼び出します。

• データ分析アプリケーションでタンブリングウィンドウとしてレコードがアプリケーション内ストリームに出力される場合、AWS Lambda 宛先関数はタンブリングウィンドウトリガーごとに呼び出されます。たとえば、60 秒のタンブリングウィンドウを使用してレコードを宛先のアプリケーション内ストリームに送信する場合、Lambda 関数は 60 秒に 1 回呼び出されます。

• レコードがアプリケーション内の宛先アプリケーション内ストリームに連続クエリまたはスライディングウィンドウとして送信される場合、Lambda 宛先関数は 1 秒に約 1 回呼び出されます。

注記

Lambda 関数呼び出しリクエストごとのペイロードサイズ制限が適用されます。これらの制限を超えると、出力レコードが分割され、複数の Lambda 関数呼び出しにわたって送信されます。

出力として使用する Lambda 関数の追加

次の手順では、Lambda 関数を Kinesis Data Analytics アプリケーションの出力として追加する方法を示します。

1. AWS Management Console にサインインし、Kinesis Data Analytics コンソール (https://console.aws.amazon.com/kinesisanalytics) を開きます。

2. リストからアプリケーションを選択し、[Application details] を選択します。

3. [Destination] セクションで、[Connect new destination] を選択します。

4. [Destination (宛先)] 項目に、[AWS Lambda function ( 関数)] を選択します。

5. [レコードの配信先AWS Lambda] セクションで、既存の Lambda 関数とバージョンを選択するか、[新規作成] を選択します。

6. Lambda 関数を新規作成する場合は、以下の操作を行います。

   1. 提供されているいずれかのテンプレートのいずれかを選択します。詳細については, アプリケーション送信先の Lambda 関数を作成.

   2. [関数の作成] ページが新しいブラウザタブで開きます。[Name (名前)] ボックスで、関数にわかりやすい名前を付けます (例: myLambdaFunction)。

   3. アプリケーションの後処理機能のテンプレートを更新します。Lambda 関数の作成については、AWS Lambda開発者ガイドの「はじめに」を参照してください。

   4. Kinesis Data Analytics コンソールの Lambda 関数リストで、先ほど作成した Lambda 関数を選択します。Lambda 関数のバージョンは [$LATEST] ($最新) を選択します。

7. [In-application stream] セクションで、[Choose an existing in-application stream] を選択します。[In-application stream name] に、アプリケーションの出力ストリームを選択します。選択した出力ストリームの結果は、Lambda 出力関数に送信されます。

8. 残りのフォームはデフォルト値のままにして、[Save and continue] を選択します。

これで、アプリケーションはアプリケーション内ストリームから Lambda 関数にレコードを送信します。デフォルトテンプレートの結果は Amazon CloudWatch コンソールで確認できます。AWS/KinesisAnalytics/LambdaDelivery.OkRecordsメトリクスを監視して、Lambda 関数に配信されているレコードの数を確認します。

よくあるLambda AS出力の失敗

Lambda 関数への配信が失敗する一般的な理由は次のとおりです。

• Lambda 関数に送信されたバッチ内のすべてのレコード (レコード ID を含む) が Kinesis Data Analytics サービスに返されるわけではありません。

• レスポンスにレコード ID、またはステータスフィールドのいずれかが欠落しています。

• Lambda 関数のタイムアウトは、Lambda 関数内のビジネスロジックを実行するのに十分ではありません。

• Lambda 関数内のビジネスロジックは、すべてのエラーをキャッチしないため、処理されない例外のためにタイムアウトとバックプレッシャーが生じます。これらのメッセージは、「ポイズンピル」と呼ばれることが少なくありません。

データ配信に失敗した場合、Kinesis Data Analytics は成功するまで同じレコードセットで Lambda 呼び出しを再試行し続けます。障害を把握するには、 CloudWatch 次の指標を監視できます。

• CloudWatch 出力指標としての Kinesis Data Analytics アプリケーション Lambda: その他の統計情報の中でも、成功数と失敗数を示します。詳細については、「Amazon Kinesis Analytics Metrics」を参照してください。

• AWS Lambda CloudWatch 関数メトリックとログ。