ScalaでKinesisのConsumerを作るチュートリアル #AWS

概要

ストリームを作り、Scalaで内容をtailするところまでを。書いているとおりに実行すれば動くことを目標にしてます。

Kinesisはざっくり言うと大規模OK Pub/Subメッセージキューという理解でいいと思います。

ストリームがPub/Subキューみたいなもので、レコードは24時間保持されます(現在は24時間以上設定で変更可能)。シャードをいくつか束ねて構成されていて、シャードを増やすことでPut/GetのQPSを増やすことが出来ます。パーティションキーによりどのシャードに入るか決まります。なのでホットシャードが起きないパーティションキーとなるように設計する必要があります。迷ったら乱数を入れるといいと思います。

偏りができて、あるシャードだけが忙しい、みたいな状況になるとシャードを増やしてもパフォーマンスを向上させることができなくなります。

Put後にシャード振り分ける仕組み

レコードのPutはストリームに対して行います。Putしたとき、パーティションキーはmd5でhash化され10進数化されます。シャードはパーティションキーのRangeを持っていて、Rangeがにあうシャードにレコードが格納されます。

Recordはデータ + パーティションキーの構造です。

Getするとき

レコードはシャードに分散されて格納されているので、Getはストリームではなくシャードに対して行います。シャードに入ったレコードはシーケンス番号を持っています。このシーケンス番号により、何処までレコードを読み込んだかを管理します。

KinesisはPull型のPub/Subです。Google Pub/Subはエンドポイントを用意すればそこにPushする仕組みを取れるらしいです。

KCLがやってくれること

何処までレコードを読み込んだのか、複数サーバーで処理するときとか・・、シャードを増やしたときとか・・、ポーリングしなきゃ？・・・なんか実装めんどくさそうです・・・が、この辺はKinesis Clinet Libraryが全部やってくれます。

チェックポイント処理
- 何処まで処理したかをDynamoDBの保存する
ワーカーの管理
- そのシャードはどのワーカーが担当しているのか
- ワーカーが死んでいたら開いてるワーカーにスイッチ

（あれ、意外と長くなってきた。）詳しいことはAmazonのドキュメントに日本語でのっています。

なので我々は「レコードをどう処理するか」を実装します。具体的にはIRecordProcessorを継承したRecordProcessorを作ります。

KCLを動かす

まずストリームを作る

# stream名をきめておく
stream_name="kinesis-test-stream"

# streamを作成する
aws kinesis create-stream --stream-name $stream_name --shard-count 1

# StreamStatusがCREATINGからACTIVEになるのを確認する。
aws kinesis describe-stream --stream-name $stream_name

scalaの環境を作ります

ストリームが作られるのを待っている間にScalaでKinesis Consumerの準備をします。

1.ディレクトリ作成

mkdir kinesis-consumer
cd kinesis-consumer
vim build.sbt

2.build.sbt作成

build.sbt

name := "kinesis-client-library-start"

version := "1.0"

scalaVersion := "2.11.7"

libraryDependencies ++= Seq(
  "com.amazonaws" % "amazon-kinesis-client" % "1.6.1"
)

3.scala
src/main/scala/Main.scala

import java.net.InetAddress
import java.util
import java.util.UUID
import com.amazonaws.internal.StaticCredentialsProvider
import com.amazonaws.auth.{BasicAWSCredentials, AWSCredentialsProvider}
import com.amazonaws.services.kinesis.clientlibrary.interfaces.{IRecordProcessorCheckpointer, IRecordProcessor, IRecordProcessorFactory}
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.{Worker, KinesisClientLibConfiguration}
import com.amazonaws.services.kinesis.clientlibrary.types.{UserRecord, ShutdownReason}
import com.amazonaws.services.kinesis.model.Record

object Main {
  val accessKeyId = System.getProperty("accessKeyId")
  val secretAccessKey = System.getProperty("secretAccessKey")

  val appName = "kinesis-test-app"
  val streamName = "kinesis-test-stream"

  val initialPosition = "LATEST"
  val region = "ap-northeast-1"
  val idleTimeBetweenReadsInMillis = 3000

  def main(args: Array[String]): Unit = {
    val workerId = InetAddress.getLocalHost.getCanonicalHostName + ":" + UUID.randomUUID
    val credentialsProvider: AWSCredentialsProvider = new StaticCredentialsProvider(new BasicAWSCredentials(accessKeyId, secretAccessKey))

    val kclConf = new KinesisClientLibConfiguration(appName, streamName, credentialsProvider, workerId)
      .withInitialPositionInStream(InitialPositionInStream.valueOf(initialPosition))
      .withRegionName(region)
      .withIdleTimeBetweenReadsInMillis(idleTimeBetweenReadsInMillis)

    val worker = new Worker(StreamTailProcessor.processorFactory, kclConf)
    println(s"worker start. name:$appName stream:$streamName workerId:$workerId")
    worker.run()

  }
}

class StreamTailProcessor extends IRecordProcessor{
  override def shutdown(checkpointer: IRecordProcessorCheckpointer, reason: ShutdownReason): Unit = {
    println(s"Shutting down record processor")
  }

  override def initialize(shardId: String): Unit = {
    println(s"Initialising record processor for shard: $shardId")
  }

  override def processRecords(records: util.List[Record], checkpointer: IRecordProcessorCheckpointer): Unit = {
    import scala.collection.JavaConversions._
    records foreach { r =>
      val line = new String(r.getData.array)
      println(s"[stream-tail] $line")
    }
  }
}

object StreamTailProcessor {
  def processorFactory = new IRecordProcessorFactory {
    def createProcessor(): IRecordProcessor = new StreamTailProcessor
  }
}

実行します

ストリームがACTIVEになったら、scalaのkinesis consumerを起動します。

1.実行します

sbt run -DaccessKeyId=XXX -DsecretAccessKey=XXX

2.ストリームにデータを投入します

aws kinesis put-record --stream-name $stream_name --data hogehoge_$RANDOM --partition-key $RANDOM

3.正座待機

kinesisのinfoログと共に以下の様な標準出力が得られました。

[stream-tail] hogehoge_18591

4.クロージング

満足したらストリームを削除します。放置すると課金されます。

aws kinesis delete-stream --stream-name $stream_name

終わり

Kinesisにデータを入れて（Producer）、処理する（Consumer）ところまで体験できました。
サンプルコードは以下においてあります。