サンプル関数コード - AWS Lambda

サンプル関数コード

Amazon Kinesis からのイベントを処理するには、イベントオブジェクト内のレコードを反復処理し、それぞれに含まれる Base64 エンコードされたデータをデコードします。

注記

このページのコードは、集計レコードをサポートしていません。Kinesis プロデューサーライブラリ設定で集計を無効にするか、Kinesis レコード集計ライブラリを使用すると、レコードの集計を解除できます。

サンプルコードは以下の言語で利用可能です。

Node.js 12.x

以下のコード例では、Kinesis イベント入力を受け取り、含まれるメッセージを処理します。例示のため、このコードでは受信イベントデータの一部を CloudWatch Logs に書き込みます。

例 index.js

console.log('Loading function'); exports.handler = function(event, context) { //console.log(JSON.stringify(event, null, 2)); event.Records.forEach(function(record) { // Kinesis data is base64 encoded so decode here var payload = Buffer.from(record.kinesis.data, 'base64').toString('ascii'); console.log('Decoded payload:', payload); }); };

サンプルコードを zip ファイルに圧縮し、デプロイパッケージを作成します。手順については、「.zip ファイルアーカイブで Node.js Lambda 関数をデプロイする」を参照してください。

Java 11

Kinesis イベントレコードデータを入力として受け取り、それを処理する Java コードの例を以下に示します。例示のため、このコードでは受信イベントデータの一部を CloudWatch Logs に書き込みます。

コードでは、recordHandler はハンドラーです。このハンドラーは、定義済みの KinesisEvent クラスを使用します。このクラスは aws-lambda-java-events ライブラリで定義されています。

例 ProcessKinesisEvents.java

package example; import com.amazonaws.services.lambda.runtime.Context; import com.amazonaws.services.lambda.runtime.RequestHandler; import com.amazonaws.services.lambda.runtime.events.KinesisEvent; import com.amazonaws.services.lambda.runtime.events.KinesisEvent.KinesisEventRecord; public class ProcessKinesisRecords implements RequestHandler<KinesisEvent, Void>{ @Override public Void handleRequest(KinesisEvent event, Context context) { for(KinesisEventRecord rec : event.getRecords()) { System.out.println(new String(rec.getKinesis().getData().array())); } return null; } }

ハンドラーが例外をスローせずに正常に戻った場合、Lambda はレコードの入力バッチが正しく処理されたと見なし、ストリーム内の新しいレコードの読み取りを開始します。ハンドラーによって例外がスローされる場合、Lambda はレコードの入力バッチが処理されていないと見なし、レコードの同じバッチで関数を再度呼び出します。

Dependencies

  • aws-lambda-java-core

  • aws-lambda-java-events

  • aws-java-sdk

Lambda ライブラリの依存関係を使ってコードを構築し、デプロイパッケージを作成します。手順については、「.zip または JAR ファイルアーカイブで Java Lambda 関数をデプロイする」を参照してください。

C#

Kinesis イベントレコードデータを入力として受け取り、それを処理する C# コードの例を以下に示します。例示のため、このコードでは受信イベントデータの一部を CloudWatch Logs に書き込みます。

コードでは、HandleKinesisRecord はハンドラーです。このハンドラーは、定義済みの KinesisEvent クラスを使用します。このクラスは Amazon.Lambda.KinesisEvents ライブラリで定義されています。

例 ProcessingKinesisEvents.cs

using System; using System.IO; using System.Text; using Amazon.Lambda.Core; using Amazon.Lambda.KinesisEvents; namespace KinesisStreams { public class KinesisSample { [LambdaSerializer(typeof(JsonSerializer))] public void HandleKinesisRecord(KinesisEvent kinesisEvent) { Console.WriteLine($"Beginning to process {kinesisEvent.Records.Count} records..."); foreach (var record in kinesisEvent.Records) { Console.WriteLine($"Event ID: {record.EventId}"); Console.WriteLine($"Event Name: {record.EventName}"); string recordData = GetRecordContents(record.Kinesis); Console.WriteLine($"Record Data:"); Console.WriteLine(recordData); } Console.WriteLine("Stream processing complete."); } private string GetRecordContents(KinesisEvent.Record streamRecord) { using (var reader = new StreamReader(streamRecord.Data, Encoding.ASCII)) { return reader.ReadToEnd(); } } } }

.NET Core プロジェクトの Program.cs を上記のサンプルに置き換えます。手順については、「.zip ファイルアーカイブを使用して C# Lambda 関数をデプロイする」を参照してください。

Python 3

Kinesis イベントレコードデータを入力として受け取り、それを処理する Python コードの例を以下に示します。例示のため、このコードでは受信イベントデータの一部を CloudWatch Logs に書き込みます。

例 ProcessKinesisRecords.py

from __future__ import print_function #import json import base64 def lambda_handler(event, context): for record in event['Records']: #Kinesis data is base64 encoded so decode here payload=base64.b64decode(record["kinesis"]["data"]) print("Decoded payload: " + str(payload))

サンプルコードを zip ファイルに圧縮し、デプロイパッケージを作成します。手順については、「.zip ファイルアーカイブで Python Lambda 関数をデプロイする」を参照してください。

Go

Kinesis イベントレコードデータを入力として受け取り、それを処理する Go コードの例を以下に示します。例示のため、このコードでは受信イベントデータの一部を CloudWatch Logs に書き込みます。

例 ProcessKinesisRecords.go

import ( "strings" "github.com/aws/aws-lambda-go/events" ) func handler(ctx context.Context, kinesisEvent events.KinesisEvent) { for _, record := range kinesisEvent.Records { kinesisRecord := record.Kinesis dataBytes := kinesisRecord.Data dataText := string(dataBytes) fmt.Printf("%s Data = %s \n", record.EventName, dataText) } }

go build を使って実行可能ファイルを構築し、デプロイパッケージを作成します。手順については、「.zip ファイルアーカイブを使用して Go Lambda 関数をデプロイする」を参照してください。