Kafka x Spark Streamingの実用例の紹介

下記で発表した資料です
Apache Kafka Meetup Japan #4 @Yahoo! JAPAN https://kafka-apache-jp.connpass.com/event/77889/?utm_campaign=event_reminder&utm_source=notifications&utm_medium=email&utm_content=title_link #kafkajp

1. Kafka x Spark Streamingの 実用例の紹介 株式会社 MicroAd 松宮 康二 1
2. 自己紹介 • 松宮 康二 (まっつーと呼ばれていますが・・) • 社会人もうすぐ3年目 • インターネット広告の配信システムを作ってます ◦ いわゆるアドテクという業界 • 主にScalaを書いてます ◦ 1.5年くらい • 会社ではスクラムマスターという 役割を担っています． • たまにQiitaに記事を書いてます ◦ https://qiita.com/mattsu6 • 最近はSparkを使ってます 2 @mattsu6666
3. 今日のテーマ Kafka x Spark Streamingでアクセスログを リアルタイムに整形・加工・蓄積してみた！ 3
4. そもそもSpark Streamingって？ • 分散ストリーム処理基盤 ◦ 無限に生成され続けるデータをリアルタイムに処理し続けること (ストリーム処理) • メッセージキューやログ収集ツールがインプットになる • 他のストリーム処理基盤 ◦ Twitter Heron, Apache Storm, Akka Streams, Kafka Streams, Apache Beamなど http://mogile.web.fc2.com/spark/spark200/streaming-programming-guide.html 4
5. 利用したケース • ビジネス要望 「色んなサイトのアクセスログをリアルタイムに分析したい！」 • アクセスログは大量に発生するのでパフォーマンス要件はシビア ◦ 数万QPSが要求される • アクセスログを整形・加工する必要がある ◦ 様々なDBとやりとりする必要がある 色々検討した結果Spark Streamingによる ストリーム処理を採用 5
6. Apache KafkaとSpark Streamingの連携 • Spark Streamingの入力ソースにはKafkaを採用 • SparkのパッケージにKafka連携が含まれているため，連携しやすい ◦ https://spark.apache.org/docs/2.2.0/streaming-kafka-0-10-integration.html • めちゃ簡単に分散ストリーム処理を実現出来ます！ 6 様々なサーバから Kafkaに一旦集める ログ収集 サーバ ログ収集 サーバ ログ収集 サーバ アクセス ログ アクセスログの重複排除・整 形・加工など
7. 2種類の連携方式 - レシーバを利用する方法 • 古い連携方法(Spark 1.3以前かつKafka0.8 ~ 0.10) • Zookeeperを介してオフセットを管理する • ブローカと直接通信する方法(DirectStream. Kafka 0.10以降が対応) 7 ドライバ エグゼキュータ エグゼキュータ レシーバ レシーバ 1. Kafkaからデータを取得 2. オフセットをZookeeperに保存 (Kafka’s high level API ) エントリポイント アプリの制御 実際に処理する プロセス
8. 2種類の連携方式 - レシーバを利用する方法 • 受け取ったデータをHDFSに書き込むことでAt least onceを実現 ◦ 障害発生時はHDFSに書き込まれたログを元に復旧 8 ドライバ エグゼキュータ エグゼキュータ レシーバ レシーバ 1. Kafkaからデータを取得 3. オフセットをZookeeperに保存 (Kafka’s high level API ) HDFS 2. 取得したデータをHDFSに保存 受信するデータと Zookeeperのオフ セットに不一致が 生じることがある
9. 2種類の連携方式 - ブローカと直接通信する方法 • 新しい連携方法(Spark 1.3以降かつKafka0.10以降) • エグゼキュータとKafkaブローカが直接通信する方法 ◦ DirectStreamと呼ばれる • 現状はexperimentalとなっており，仕様変更が発生する可能性有り 9 9 ドライバ エグゼキュータ エグゼキュータ 1. Kafkaからデータを取得 定期的にオフセットを通知 2. オフセット情報を通知 Zookeeperを利用しないため オフセットの不一致が発生しない！ Exactly onceを実現できる！ オフセットはスパーク内で 管理
10. DirectStreamのメリット • KafkaのDirectStreamを用いると分散ストリーム処理が実装しやすい • KafkaのパーティションとSparkのエグゼキュータが自動的に1対1で紐づく 10 パーティション1 パーティション2 パーティション3 エグゼキュータ エグゼキュータ エグゼキュータ Kafkaのパーティションの 数だけ並列数が増加する ため構成がシンプル！ デメリットとしてはZookeeperにオフセットを記録しないので Zookeeperベースの 監視ツールが機能しなくなる模様・・
11. DirectStreamのメリット • さらに，パーティション毎に独立して処理ができるため，シャッフルを発生を最小限 に抑えられる 11 パーティション1 パーティション2 パーティション3 エグゼキュータ エグゼキュータ エグゼキュータ サイトAの アクセスログ サイトBの アクセスログ サイトCの アクセスログ 例えばサイト毎に時系列を担保して ログを出力したいとか 過去のアクセスログを参照して処理 (ウィンドウ集計処理)したい場合に 有効
12. DirectStreamでない場合・・ • パーティションとエグゼキュータをマッピングしていないので 自分で作り込む必要がある 12 パーティション1 パーティション2 パーティション3 エグゼキュータ サイト毎に分割する処 理が必要 (groupBy) エグゼキュータ エグゼキュータ エグゼキュータ
13. 実装例 13 val sparkConf = new SparkConf().setMaster("local").setAppName("test-spark") val ssc = new StreamingContext(sparkConf, Seconds(10)) val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "localhost:9092", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], "enable.auto.commit" -> (false: java.lang.Boolean) ) val topics = Array("random") val kafkaStream = KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams)) // アクション kafkaStream.foreachRDD { rdd => // パーティション毎にループする // 何か処理を書く } ssc.start() ssc.awaitTermination() Kafkaのパーティション数 に応じて勝手に 分散処理します
14. おわりに • KafkaとSpark Streamingを利用してアクセスログの整形・加工・蓄積を やってみました(本LTではKafkaとSpark Streamingの連携方法を説明) • Spark StreamingでKafkaのDirectStreamを活用すると分散ストリーム処理が書き やすくなります • 明日はScala Matsuriに行ってきます！ • 会社にScala書く人が少なく寂しいのでScalaのお友達が欲しい... 14