サンプル関数コード

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 はレコードの入力バッチが処理されていないと見なし、レコードの同じバッチで関数を再度呼び出します。

依存関係

• 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 ファイルに圧縮し、デプロイパッケージを作成します。手順については、「Python Lambda 関数で .zip ファイルアーカイブを使用する」を参照してください。

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 関数をデプロイする」を参照してください。