Java Clientで入門する Apache Kafka #jjug_ccc #ccc_e2

Java Clientで入門する Apache Kafka JJUG CCC 2017 Spring 2017年5月20日 ヤフー株式会社 データ＆サイエンスソリューション統括本部 データプラットフォーム本部 開発4部 パイプライン 1 森谷 大輔 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

自己紹介 • 森谷 大輔 • インフラエンジニア • • どちらかというとBEアプリケーション開発 ヤフーのデータパイプラインをつくったり運用したり • 対外発表 • ストリーム処理プラットフォームにおけるKafka導入事例 (Apache Kafka Meetup Japan #1: 2016) • Hadoop Summit 2016 @San Jose ストリーム処理関連の報告 (Stream Processing Casual Talks #1: 2016) • Kafka 0.10.0アップデート プロダクション100ノードでやってみた (D&S Data Night vol.03: 2016) • 噛み砕いてKafka Streams (Apache Kafka Meetup Japan #2: 2016) • Elasticsearch 5.2とJava Clientで戯れる (第18回Elasticsearch勉強会: 2017) 2 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

ゴール（と注意） Kafkaをこれから始めたい、始めたばか りの人向けにKafkaの基本を実演ベース で押さえて帰ってもらう • ※既にガンガン使ってるぜという人は退屈 かも、でもちょっとはいい情報もあるかも • ※現行最新のバージョン0.10.2.0を前提 3 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

アジェンダ • • • • 4 Kafkaとは？ Kafkaの基本 Java ClientでKafkaを体感する まとめ Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Kafkaとは？ 時間を割きます 5 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Apache Kafka スケーラブルな分散pub/sub型 メッセージングシステムを実現するために LinkedInが開発してOSS化した (2011年~) Apacheトップレベルプロジェクトで配布（2012年~） 6 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

実績 Fortune 500の内 実に3分の1がKafkaを 利用しているらしい 7 https://kafka.apache.org/powered-by から一部を抜粋 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

• https://kafka.apache.org/powered-by

Kafka Summit • Confluent, inc.が主催 • SF (2016/4, 2017/8), NY (2017/5) 8 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

ユーザ会 • Apache Kafka JP • 9 https://kafka-apache-jp.connpass.com/ https://kafka.apache.org/events Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

• https://kafka-apache-jp.connpass.com/
• https://kafka.apache.org/events

Apache Kafka スケーラブルな分散pub/sub型 メッセージングシステムを実現するために LinkedInが開発してOSS化した (2011年~) Apacheトップレベルプロジェクトで配布（2012年~） 10 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

publish/subscribeモデル • Wikipediaには出版-購読型モデルとある • データの送受信のかたち publisher subscriber publisher 仲介者 publisher • subscriber subscriber データの送信者(publisher)は受信者(subscriber)を気にしないでどんどん 送る 受信者は送信者を気にせず自分のタイミングで取る • 11 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

publish/subscribeモデル 疎結合化 PUSH PULL publisher subscriber publisher 仲介者 publisher • • subscriber つまりpublisherとsubscriberを非同期に分離した もしPUSH, PUSHだとどんな問題がある？ 12 subscriber Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

具体的にはなんの用途で使うの？ • 2つ • データパイプライン • ストリーム処理 と公式docのトップにもある 13 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

データパイプライン Data Source RDBMS Data Source Data Source 14 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

データパイプライン ストリーム処理 Data Source Data Source Data Source RDBMS Data Source Data Source Data Source 15 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

データパイプライン ストリーム処理 Data Source Data Source Data Source RDBMS データレイク・・・ Data Source マイクロサービス・・・ Data Source Data Source 16 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

データパイプライン ストリーム処理 Data Source Data Source RDBMS Data Source Data Source Data Source 17 Data Source Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

ストリーム（リアルタイム）処理 • 課題： このデイリーバッチが終わる明朝まで分析がで きないよ・・・ • ASAPの意思決定 18 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

ストリーム（リアルタイム）処理 • • モジュールの日次クリック率推移を知りたい モジュールのクリック率急増を知りたい • • 遅延が多い路線を知りたい 電車の遅延を知りたい • • 2016年に話題になったツイートワードを知りたい 今見ているテレビ番組のバズツイートワードを知りたい • • ABテストの分析をしたい 非常に悪い結果を出しているバケットテストをすぐに停止したい • • マシンの故障率を知りたい マシンが故障しそうになったら知りたい • • ユーザの平均的な興味から広告を出したい ユーザが今だけ興味を持っている広告を出したい 19 変化の通知が、実際に変化した点から時 間が経つに連れて価値が低下する バッチ ストリーム 高価値 数秒〜10分 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved . 価値の低下

ストリーム（リアルタイム）処理 • ストリーミングデータ • 終わりがない、連続的に発生するデータ • を処理するためにはキューイングの機能が必要 • 類似：ActiveMQ, RabbitMQ, ZeroMQ, Redis • https://www.slideshare.net/laclefyoshi/ss-67658888 20 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

• https://www.slideshare.net/laclefyoshi/ss-67658888

Kafkaは • データパイプラインをシンプルにする • ストリーミングデータを扱える を両方実現できるように設計された • 特徴 • バッチ・リアルタイム分析共にサポート • 受け取ったレコードは直ちに永続化 • スケーラビリティ • • 21 デイリー1兆4千万のレコードを捌く@LinkedIn フォールトトレラント Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Kafkaの基本 22 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

pub/sub publisher subscriber publisher 仲介者 subscriber publisher 23 subscriber Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Kafkaを構成する4コンポーネント Producer Consumer Producer Consumer Broker Consumer Producer 狭義にはこれだけを Kafkaと言うこともある 24 Co p yrig ht © 2 0 1 7 ZooKeeper Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Kafkaを構成する4コンポーネント • • • • 25 Producer Consumer Broker ZooKeeper Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

ZooKeeper • Apache ZooKeeper • 可用性と信頼性の高い分散コーディネーションシステム （詳細は割愛） • Hadoop等でも使われるOSS • Kafkaでは↓のために使われる • クラスタマネジメント • 死活管理 • ACL情報のストア 26 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Message Producer Consumer Producer Consumer Broker Consumer Producer Message（＝Record） • Kafka独自フォーマットのバイナリ • keyとvalueとtimestamp等のメタデータで構 成 • key, valueはユーザ任意の型 27 Co p yrig ht © 2 0 1 7 ZooKeeper Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Topic Producer Consumer Broker weblog Producer Consumer purchase customer Producer 28 Topic • Messageストリームのラベル • ユーザが名付ける • Producerは1個か複数のtopicにMessageを投げ る Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved . Consumer

Partition weblog: partition partition • Topicはさらに負荷分散のた めにPartitionに分割される weblog: partition • Partition数はTopic毎にユー ザが決める purchase: Producer purchase: partition • Partition数は増やせるが減ら せない weblog: partition purchase: 29 Co p yrig ht © 2 0 1 7 partition Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Producer: key partitioning weblog: partition purchase: Producer partition weblog: partition purchase: partition weblog: partition purchase: 30 Co p yrig ht © 2 0 1 7 • keyのハッシュ値から投入先 partitionを決定する • 同じkeyのMessageは同じ partitionに投入される • keyを指定しなければラウンド ロビン • またはpartition割り振りロジッ クをユーザがカスタマイズす る partition Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Log weblog: partition purchase: partition • partitionは任意のディレクトリ パスにファイルとして永続化 される • これをLogという Producer weblog: partition purchase: partition weblog: partition purchase: 31 Co p yrig ht © 2 0 1 7 • 紛らわしいことにKafkaそのも ののシステムログも別に存在 する、がそれではない • 保持期間が指定できる partition Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Consumer Group Consumer Broker Producer purchase Consumer group.id=shopping Consumer • グルーピングされたconsumerは他のconsumerを 気にせずtopicのMessageを漏れなく分散 consumeする 32 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved . group.id=marketing

Offset ということはConsumer Group毎に どこまで読んだかをConsumerが覚えている Consumer partition 0 0 1 Consumer group.id=shopping 2 partition 1 Consumer 0 1 group.id=marketing 33 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Offset • 「group.id=shoppingとしてtopic=purchaseのpartition=0のoffset=2まで読ん だ」 • この情報はKafkaの専用topicに書き込まれる Consumer Offsetは一度決まったら変わらない Consumer partition 0 0 1 group.id=shopping 2 partition 1 Consumer 0 1 group.id=marketing 34 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Java ClientでKafkaを体 感する 35 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Java Client • Producer, Consumer APIとしてJava Client が公式に用意されている • しかし3rdパーティ製で他言語でも使える （C/C++, Go, Python等） • 特にConsumerはアプリケーションコードを書 くことが多い 36 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Consumer Client • • 特にConsumerはアプリケーションコードを書くことが 多い よりハイレベルなAPIやフレームワークもある Kafka Streams • 37 Java Clientは最もローレベルなAPIともいえる Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

ハイレベルAPIがあるのにわざわざ ローレベルAPIの使い方を見る必要がある？ ある • • • ハイレベルAPIを使いこなすのは簡単でない、Java Clientを把握 していることはハイレベルAPIを使いこなす上でとても有用 フレームワークは構築から試すまでがかなり大変、ちょっとした アプリやテストならJava Clientが一番低コストだったり ローレベルといっても直感的でないわけではない • 38 （旧APIのSimpleConsumerさんとはわけが違う） Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

これから試すこと • Kafkaアプリケーションの全ての基本 • サンプルデータをProducerがBrokerにproduce する • ConsumerがBrokerからそのデータをconsume して確認する Producer k: null, v: “こんにちは世界” 39 Consumer Broker Co p yrig ht © 2 0 1 7 “こんにちは世界” ? Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

どうやって動作確認する？ • Producer, Consumer, Broker, ZooKeeper 1. 2. 3. 4. ZooKeeper, Kafkaを開発機にインストール ZooKeeper daemonとBroker daemonを起動して お試しProducerコードを書いてjarを配備して起動して お試しConsumerコードを書いてjarを配備して起動して 40 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

面倒！ • フレームワークだともっと面倒 41 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

これがあれば一連の流れが試せる 4つの最低限コンポーネント • • • • ZooKeeper: JVMで動く Kafka Broker: JVMで動く Kafka Producer: JVMで動く Kafka Consumer: JVMで動く • Javaコードペラ一枚でできるのでは？ 42 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

A. できる • つまりJUnit上で試せる • Kafkaアプリケーションを開発するときはこ の手を使ってテストすると良い • ConsumerがStormやKafka Streamsだった としても、それもJVMなので同じ手が使える 43 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

• ZooKeeperとBrokerをJavaコード上で起動する方法は公 式docのFAQにある • https://cwiki.apache.org/confluence/display/KAFKA/FAQ#FAQ-Unittesting • やや長いのでライブラリを作って一行くらいで起動できる ようにしておくと吉 • 今回はすみません、自作のものを使います • 44 使い勝手は違いますが基本やっていることはFAQと同じ Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

• https://cwiki.apache.org/confluence/display/KAFKA/FAQ#FAQ-Unittesting

Java Client【とJUnit】 でKafkaを体感する （今度こそ） 45 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

これから試すこと（再掲） • Kafkaアプリケーションの全ての基本 • サンプルデータをProducerがBrokerにproduce する • ConsumerがBrokerからそのデータをconsume して確認する Producer k: null, v: “こんにちは世界” 46 Consumer Broker Co p yrig ht © 2 0 1 7 “こんにちは世界” ? Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

デモ1 • • • pom テストコード broker partitionの実体を見てみましょう • Producer, ConsumerのJavaDoc • https://kafka.apache.org/0102/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html • https://kafka.apache.org/0102/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html 47 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

• https://kafka.apache.org/0102/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html
• https://kafka.apache.org/0102/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html

Consumer Groupを体感する • Aというgroup.idとBというgroup.idで同じ topicのMessageを読んでみる Broker Producer Consumer group.id=A test-topic Consumer group.id=B 48 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

デモ2 49 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

• 5件投げたら5件ずつ、合計10件読 めた • consumer group BをAに変えてもう 一回試すとどうなるか 50 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

デモ2.1 51 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

• 1スレッドだけが5件読んでもう1スレッドは働 かなかった • なぜかは後で • とりあえずconsumer groupを分ければ別とし て読めることがわかった 52 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

partitioningで負荷分散を体感する • • • • • 53 topic毎にpartition数を指定できる partition=1だとBroker nodeを何台増やしても拡張しない 1partitionは1Consumerからしか読まれない 1partitionを2Consumerから同時に読むことはできない、 2partitionなら可能 さっき1スレッドだけが5件読んでもう1スレッドが働かな かった理由はそれ Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

デモ3 • partition数を2にしてさっきのコードを実行して みます • broker partitionの実体も変わっていることを確 認 54 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

今回は同プロセス内スレッドで分散さ せたが、本番ではnode間で分散され ることになるはず 55 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

key partitioning • • Aさんのログイン失敗が30分以内に5回以上を検知 Kafka Consumerで実装した場合 partition 0 Consumer 56 partition 1 2回失敗 ？ Consumer Co p yrig ht © 2 0 1 7 partition 2 2回失敗 ？ Yaho o Jap an Co rp o ratio n. All Rig hts Reserved . Consumer 1回失敗 ？

こんなときに使える • keyでgroupingして集約したい • keyでストリームデータ同士をjoinしたい • uniqueKeyを入れてウィンドウで重複除 去したい 57 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

デモ4 • Aという文字列をkeyに入れてさっきの 2partitionに投げてみます • さっきは分散したが今回は？ 58 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

デモ5 • 圧縮 • するとしないのとではネットワークトラフィック、 Brokerのディスク消費が大きく変わる • • • 59 その代わりCPUを食う gzip, snappy, lz4 Consumerで読むときはなにで圧縮されたか気 にする必要は無い Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

マルチスレッド実装の注意点 • • テストのためにConsumerはマルチスレッドで試してきた 本番でも性能を出すために恐らくマルチスレッド実装する • • KafkaProducerクラスはスレッドセーフ KafkaConsumerクラスはスレッドセーフではない • Consumerオブジェクトをスレッド間で共有して操作すると ConcurrentModificationException 各スレッドで作るようにすればOK • 60 （※wakeupメソッドだけ例外） Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

KafkaConsumer 0.10.2で使える便利メソッド紹介 返り値型 void メソッド • • • 機能、使いどころ commitSync() commitSync(Map<TopicPartition, OffsetAndMetadata> offsets) commitAsync(OffsetCommitCallback callback) Offsetのコミットを自動定期に任せない場合、自分でOffset コミットしたいタイミングを指定する。 特定のpartitionについて特定のOffsetでコミット、ということ もできる。 処理を含めてAt-least-onceを実現する場合は必須。 Map<TopicPar tition, OffsetAndTim estamp> • offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch) 最近追加された強力な機能。 MessageのtimestampからOffsetを逆引きする。 下記のseekと合わせて使うことで何時何分のデータから何 時何分のデータを再度consumeし直すといったユースケー スを実現できる。 void • • • seek(TopicPartition partition, long offset) seekToBeginning(Collection<TopicPartition> partitions) seekToEnd(Collection<TopicPartition> partitions) 今読んでいるOffsetを途中で変える機能。 ほとんどのConsumerアプリケーションで稼働中に任意に Offsetを変えたい要望が出てくるので重要度が高いメソッド。 void • • pause(Collection<TopicPartition> partitions) resume(Collection<TopicPartition> partitions) 特定のpartitionのconsumeを一時停止したり再開させたり する。「今だけ特定のpartitionのconsumeに集中させたい」 といった要件を実現する。 便利だが実装がやや難しい。 61 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

まとめ 62 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

このセッションで • • • • • 63 どんなときにKafkaを使うといいかわかった Kafkaを試すために最低限必要なコンポーネントを覚えた JUnitでKafkaを試せることがわかった Java ClientでKafkaのpartitioning, Consumer Groupの機 能を体感した マルチスレッドConsumerアプリケーションを書く上で最低 限注意すべきことがわかった Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

Kafka Meetup JP #3 7/6（木）開催 会場は LODGE@ヤフー 以下connpassで案内予定 https://kafka-apache-jp.connpass.com/ 64 Co p yrig ht © 2 0 1 7 Yaho o Jap an Co rp o ratio n. All Rig hts Reserved .

• https://kafka-apache-jp.connpass.com/