AWS Kinesis Client Libraryでコンシューマ開発 のバックアップ(No.4) - 闘うITエンジニアの覚え書き

aws kinesis create-stream --stream-name myFirstStream --shard-count 1

※サンプルのストリーム名から変える場合は、AmazonKinesisApplicationSample.java で定義されているストリーム名を書き換える。（後述参照）

# プロジェクトフォルダを作成
mkdir KclSample && cd KclSample

# 初期化
gradle init --type java-application

build.gradle の編集

plugins {
    id 'java'
    id 'application'
}

mainClassName = 'AmazonKinesisApplicationSample'

dependencies {
    compile 'com.google.guava:guava:23.0'
    compile 'com.amazonaws:amazon-kinesis-client:1.9.0'
    compile 'com.google.code.gson:gson:2.8.2'
    compile group: 'org.apache.logging.log4j', name: 'log4j-api', version: '2.11.0'
    compile group: 'org.apache.logging.log4j', name: 'log4j-core', version: '2.11.0'

    testCompile 'junit:junit:4.12'
}

repositories {
    jcenter()
}

jar {
    manifest {
        attributes 'Main-Class': mainClassName
    }   
} 

EclipseからMavenプロジェクトを作成し、amazon-kinesis-clientとgsonを依存ライブラリとして追記するだけ。

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>KclSample</groupId>
  <artifactId>KclSample</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <dependencies>
    <dependency>
      <groupId>com.amazonaws</groupId>
      <artifactId>amazon-kinesis-client</artifactId>
      <version>1.9.0</version>
    </dependency>
    <dependency>
      <groupId>com.google.code.gson</groupId>
      <artifactId>gson</artifactId>
      <version>2.8.2</version>
    </dependency>
  </dependencies>
</project>

https://github.com/aws/aws-sdk-java/tree/master/src/samples/AmazonKinesis にあるサンプルから必要なファイルを取得する。

# src, test フォルダをキレイにしておく
rm -rf src/main/java/*
rm -rf src/test/java/*

# 必要なサンプルソースをダウンロード
wget https://raw.githubusercontent.com/aws/aws-sdk-java/master/src/samples/AmazonKinesis/AmazonKinesisApplicationRecordProcessorFactory.java \
    -O src/main/java/AmazonKinesisApplicationRecordProcessorFactory.java
wget https://raw.githubusercontent.com/aws/aws-sdk-java/master/src/samples/AmazonKinesis/AmazonKinesisApplicationSample.java \
    -O src/main/java/AmazonKinesisApplicationSample.java
wget https://raw.githubusercontent.com/aws/aws-sdk-java/master/src/samples/AmazonKinesis/AmazonKinesisApplicationSampleRecordProcessor.java \
    -O src/main/java/AmazonKinesisApplicationSampleRecordProcessor.java

サンプルは少し古いKCLのバージョン用のものになっているので一部変更する。
また、IRecordProcessor バージョン1用のサンプルなので、バージョン2用を試したい場合は、必要に応じて書き換える。
https://docs.aws.amazon.com/ja_jp/streams/latest/dev/kinesis-record-processor-implementation-app-java.html#kcl-java-interface-v2

AmazonKinesisApplicationSample.java の修正

    .
    .

    public static final String SAMPLE_APPLICATION_STREAM_NAME = "myFirstStream";     // 別のストリーム名にする場合は変える

    .
    .

    public static void main(String[] args) throws Exception {
          .
          .
        kinesisClientLibConfiguration.withInitialPositionInStream(SAMPLE_APPLICATION_INITIAL_POSITION_IN_STREAM);
        kinesisClientLibConfiguration.withRegionName("ap-northeast-1");      // [追加] 東京リージョンを指定

          .
          .

    public static void deleteResources() {
        // Delete the stream
        AmazonKinesis kinesis = AmazonKinesisClientBuilder.standard()
            .withCredentials(credentialsProvider)
            .withRegion("ap-northeast-1")                            // [変更] 東京リージョンに変更
            .build();

        System.out.printf("Deleting the Amazon Kinesis stream used by the sample. Stream Name = %s.\n",
                SAMPLE_APPLICATION_STREAM_NAME);
        try {
            kinesis.deleteStream(SAMPLE_APPLICATION_STREAM_NAME);
        } catch (ResourceNotFoundException ex) {
            // The stream doesn't exist.
        }   

        // Delete the table
        AmazonDynamoDB dynamoDB = AmazonDynamoDBClientBuilder.standard()
            .withCredentials(credentialsProvider)
            .withRegion("ap-northeast-1")                           // [変更] 東京リージョンに変更
            .build();
        System.out.printf("Deleting the Amazon DynamoDB table used by the Amazon Kinesis Client Library. Table Name = %s.\n",
                SAMPLE_APPLICATION_NAME);
        try {
            dynamoDB.deleteTable(SAMPLE_APPLICATION_NAME);
        } catch (com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException ex) {
            // The table doesn't exist.
        }   
    }   
</pre>

AmazonKinesisApplicationSampleRecordProcessor.java の修正

.
.

//import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownReason;                   // [コメントアウト] 
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;              //  [追加] 新しいバージョンだとパッケージが変わっている

import com.google.gson.Gson;  // [追加]
import java.util.Map;  // [追加]

    .
    .

    // テキスト文字列 または Json文字列 を受け取ってログ出力するように変更。
    private void processSingleRecord(Record record) {

        String data = null;
        try {
            data = decoder.decode(record.getData()).toString();
            try {
                Map receiveJson = new Gson().fromJson(data, Map.class);
                LOG.info("##### received json : " + receiveJson + " #####");
            } catch (Exception e) {
                LOG.info("##### Received string: " + data + " #####");
            }   

        } catch (NumberFormatException e) {
            LOG.info("Record does not match sample record format. Ignoring record with data; " + data);
        } catch (CharacterCodingException e) {
            LOG.error("Malformed data: " + data, e); 
        }   
    }   

gradle run

aws kinesis put-record --stream-name myFirstStream --partition-key 123 --data "test123"
aws kinesis put-record --stream-name myFirstStream --partition-key 123 --data "{\"var1\": \"test123\"}"

※受信データがログ出力されているか確認する。

ddl/KinesisReceivedData.json

{
    "TableName": "KinesisReceivedData",
    "AttributeDefinitions": [
        { "AttributeName": "id", "AttributeType": "S" }
    ],  
    "KeySchema": [
        { "AttributeName": "id", "KeyType": "HASH" }
    ],  
    "ProvisionedThroughput": { "WriteCapacityUnits": 1, "ReadCapacityUnits": 1 }
}

AWS CLI でテーブル作成

aws dynamodb create-table --cli-input-json file://ddl/KinesisReceivedData.json

AWS Java SDK を使用して、先ほど作成したテーブルに受信データを登録するように処理を変更する。
※参考: https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/index.html
※関連: AWS Java SDKでDynamoDBのCRUDを書いてみる

AmazonKinesisApplicationSampleRecordProcessor.java

.
.

import com.google.gson.Gson;

import java.time.format.DateTimeFormatter;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
import com.amazonaws.services.dynamodbv2.document.DynamoDB;
import com.amazonaws.services.dynamodbv2.model.AttributeValue;
import com.amazonaws.services.dynamodbv2.model.PutItemRequest;
import com.amazonaws.services.dynamodbv2.model.PutItemResult;

.
.
public class AmazonKinesisApplicationSampleRecordProcessor implements IRecordProcessor {

    .
    .
    private AmazonDynamoDB amazonDynamoDB = AmazonDynamoDBClientBuilder.standard().build();                  // DynamoDBアクセス用のクライアント
    .
    .

    private void processSingleRecord(Record record) {

        String data = null;
        try {
            data = decoder.decode(record.getData()).toString();
            try {
                Map receiveJson = new Gson().fromJson(data, Map.class);
                LOG.info("##### received json : " + receiveJson + " #####");
            } catch (Exception e) {
                LOG.info("##### Received string: " + data + " #####");
            }   

            String tableName = "KinesisReceivedData";

            String id = UUID.randomUUID().toString();
            String datetime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("uuuu-MM-dd'T'HH:mm:ss.SSS"));

            Map<String,AttributeValue> item = new HashMap<String,AttributeValue>();
            item.put("id"      ,   new AttributeValue().withS(id));
            item.put("datetime",   new AttributeValue().withS(datetime));
            item.put("data"    ,   new AttributeValue().withS(data));

            PutItemResult result = amazonDynamoDB.putItem(new PutItemRequest(tableName, item));
            System.out.println(result);

        } catch (NumberFormatException e) {
            LOG.info("Record does not match sample record format. Ignoring record with data; " + data);
        } catch (CharacterCodingException e) {
            LOG.error("Malformed data: " + data, e); 
        }   
    }   

AmazonKinesisApplicationSample.java を実行する。

gradle run

testdata.json

{
  "var1":  "ABCDEFG",
  "timeStamp": "2018-02-12 18:50:05",
  "data": { "data1": "11111", "data2": "22222" }
}

AWS CLI からKinesisにレコード出力

aws kinesis put-record --stream-name myFirstStream --partition-key 123 --data file://testdata.json

※受信データがログ出力されているか確認する。

aws dynamodb scan --table-name KinesisReceivedData