Apache Kafka Meetup Japan #5 @LINE

12/18は「Apache Kafka Meetup Japan #5 @LINE」に参加してきました。

kafka-apache-jp.connpass.com f:id:kabukawa:20181216102332p:plain:w300

会場はLINE株式会社のイベントスペース。

f:id:kabukawa:20181219000700j:plain:w700 f:id:kabukawa:20181219000847j:plain:w350 f:id:kabukawa:20181219000910j:plain:w350

初めて行ったのですが、メチャ広くてキラキラなスペースですね！

会場、入れました！ #kafkajp pic.twitter.com/rjtkZrpubA
— kabukawa (@kabukawa) 2018年12月18日

たくさんのノベルティ、頂きました！ありがとうございます！なんとTシャツも！ #kafkajp pic.twitter.com/3b7taXG0LZ
— kabukawa (@kabukawa) 2018年12月18日

Apache Kafka Meetup Japan とは
スケジュール
内容
懇親会
- LT1：Custom management apps for Kafka
- LT2：prometheus-kafka-consumer-group-exporterを使った監視の話"
まとめ

Apache Kafka Meetup Japan とは

Apache kafkaに関する日本のユーザコミュニティです。

Apache Kafka Meetup Japan の勉強会への参加は2度目です。前回のイベントはこちら(ブログは書いてません)

kafka-apache-jp.connpass.com

スケジュール

時間	内容
18:45～19:30	入場受付
19:15～19:20	オープニング
19:20～20:05	"Real Time Stream Processing with KSQL and Kafka" Mr.David Peterson
20:05～20:20	"LINE Ads Platformの開発を支えるKafka" Hiromu Ogawa
20:20～20:35	"I/O intensiveなKafka ConsumerアプリケーションのスループットをLINE Ads Platformではどのように改善したか" 岡田遥来ジャコ(@ocadaruma)
20:35～20:50	"Yahoo! JAPANのKafka Platformで起きた障害とチューニング" TakumaTachibana
20:50～21:05	懇親会
20:50～20:55	"Custom management apps for Kafka" kimutansk
20:55～21:00	"prometheus-kafka-consumer-group-exporterを使った監視の話" yosshi_
21:00～21:35	LT予備枠
21:35～22:00	closingと完全撤収

内容

トゥギャられていませんが、ハッシュタグの検索結果で流れは追えると思います。 twitter.com

そしてお約束(ぇ

飲み物、頂きました。生🍺もあるみたいです（笑） #kafkajp pic.twitter.com/nsdAY7PXhN
— kabukawa (@kabukawa) 2018年12月18日

Real Time Stream Processing with KSQL and Kafka

David Peterson さん

Real Time Stream Processing with KSQL and Kafka from David Peterson

www.slideshare.net

スライドが英語なのでGoogle翻訳で適当に訳したものを貼り付けておきます。雰囲気だけでもつかめるかと。

KSQLとKafkaによるリアルタイムストリーム処理

KSQLとリアルタイムのストリーム処理

DAVID PETERSON
システムエンジニア - Confluent APAC @davidseth

Kafkaのアーキテクチャの変更？
KSQLによるストリーム処理
本番運用におけるKSQL

Confluentについての簡単な紹介
- Apache Kafkaの創設者によって設立された
- 2014年9月設立
- LinkedInで開発された技術
- アクティブなカフカ・コミットの69％

Kafkaの76％のコードはConfluentチームが作成

アーキテクチャの変更

イベント
- 販売
- 請求
- 取引
- 顧客体験

アーキテクチャの変更
私たちは古い習慣に取り組んでいます...
ビッグデータ
- より多いほうが良い
   データの価値 - データ量
ストリームデータ
- より速いほうが良い
   データの価値- データの世代

アーキテクチャの変更
私たちは古いアーキテクチャに取り組んでいます...
Lambda
- 大きい もしくは 速い
  DBストリーム - スピードテーブル
  Hadoop           - バッチテーブル

kappa
- 大きい かつ 速い
  KSQLストリーム - Kafka
  トピックA           - Kafka
    Kafka -  HDFS
    Kafka - Cassandra
    Kafka - Elastic
    Kafka - マイクロサービス

思考の変化...
Kafka：
イベント中心の思考

思考の変化...
Kafka：
エンタープライズイベント駆動
- すべてがイベントです
- 会社内のすべてのアプリケーションに即座に利用可能
- 到着時にデータを照会する機能と、遅すぎる場合にデータを照会する機能
- 単一のプラットフォームを導入してデータアーキテクチャを簡素化する可能性は何ですか？

Apache Kafka

無限のデータ保持コンピューティングと無制限のストリーミング・データをリアルタイムで大量にスケーラブルに分散、フォールト・トレラントなパブリッシュ＆サブスクライブのキー/バリュー・データストアです。

それで、Kafkaは本当は何ですか？

それは3つの主要なプリミティブ

発行と購読
ストア
プロセス

Connect API
- Kafkaと他のシステムとの信頼性と拡張性の高い統合 - コーディングは不要です。
プロデューサー＆コンシューマーAPI
- 多数の言語のオープンソースのクライアントライブラリ。システムとの直接的な統合。
ストリームAPI
- 低レベルおよびDSLは、リアルタイムでデータを処理するためのアプリケーションとマイクロサービスを作成します。

Apache KafkaとConfluentの歴史
2012年 0.7
2013年
2014年 0.8 クラスタ内レプリケーション
2015年
2016年 0.9 データ統合 0.10ストリーム処理
2017年 0.11 正確に一度のセマンティクス
2018年 1.0 1つの<dot>リリース！ CP 4.1 KSQL GA 2.0

プロデューサー
Kafka クラスター
コンシューマー

それで、ストリームは正確には何ですか？

1. TOPIC

2. STR

3. TABLE

変更履歴ストリーム - 不変イベント

元のテーブルを再構築する

ストリーム処理

標準アプリ
別のクラスタを作成する必要はありません。

あなたのアプリケーション内部に存在するストリーム処理

ストリームがテーブルを満たす
例
 - アリスがかつて行ったことのあるすべての場所
必要なときに…
 - キーのすべての値
あなたはカフカの話題を
 -  KStream
そのトピックはaと解釈されます
 - レコードストリーム
メッセージは次のように解釈されます。
 -  INSERT（追加）

ストリームがテーブルを満たす
例
 - アリスがかつて行ったことのあるすべての場所
 - アリスは今どこにいるの？
必要なときに…
 - キーのすべての値
 - キーの最新値
あなたはカフカの話題を
 -  KStream
 -  KTable
そのトピックはaと解釈されます
 - レコードストリーム
 - チェンジログ・ストリーム
メッセージは次のように解釈されます。
 -  INSERT（追加）
 -  UPSERT（既存の上書き）

同じデータですが、異なるユースケースでは異なる解釈が必要です
ユースケース1：頻繁な旅行者ステータス？
「アリスはSFO、NYC、リオ、シドニー、北京、パリ、そして最終的にはベルリンに行った。
ユースケース2：現在の場所？ KStream KTable
 -  "アリスは現在SFO、NYC、リオ、シドニー、北京、パリ、ベルリンにあります。"

KSQL

KSQL  - ストリーム処理の高速化
Kafka（データ）
  ↑ネットワークの書き込みを読む↓
KSQL（処理）
"CREATE STREAM CREATE TABLE SELECT ...など..."
必要なのはKafkaだけです - ストリーム処理用の別注システムを複雑に配置する必要はありません！

ソースコードの配布は不要
- ゼロ、まったく、1つの小さなファイル
Kafka Streamsのすべての機能はすぐに使用できます
- ただ一つのセマンティクス
- ウィンドウ処理
- イベント時の集計
- 遅れて到着するデータ
 - 分散、フォールトトレラント、スケーラブル、... KSQLの概念

KSQL  -  SELECT文の構文
"SELECT` select_expr` [、...]
  FROM `from_item` [、...]
  [ウィンドウ `window_expression`]
  [WHERE `condition`]
  [GROUP BY `グループ化式`]
  [HAVING `having_expression`]
     [LIMIT n]
from_itemは次のいずれかです。
    stream_or_table_name [ [ AS ] alias]
    from_item LEFT JOIN from_item ON join_condition

KSQLのいくつかのユースケースは？

データの探索
Kafkaのデータを簡単に検査する方法
"SHOW TOPICS;
PRINT 'my-topic' FROM BEGINNING;"
"SELECT page, user_id, status, bytes
 FROM clickstream
 WHERE user_agent LIKE 'Mozilla/5.0%';"

データの豊富化
さまざまなソースからのデータを結合して、完全な画像を表示する
"CREATE STREAM enriched_payments AS
 SELECT payment_id, u.country, total
 FROM payments_stream p
 LEFT JOIN users_table u
           ON p.user_id = u.user_id; Stream-table join"

ストリーミングETL
移動中のデータのフィルタリング、クレンジング、処理
"CREATE STREAM clicks_from_vip_users AS
 SELECT user_id, u.country, page, action
 FROM clickstream c
 LEFT JOIN users u ON c.user_id = u.user_id
   WHERE u.level ='Platinum';"

異常検出
リアルタイムでパターンや異常を特定するためのデータ集約
"CREATE TABLE possible_fraud AS
 SELECT card_number, COUNT(*)           --- 集計データ
 FROM authorization_attempts
 WINDOW TUMBLING (SIZE 5 MINUTE) --- … 5分ごとのウィンドウ
 GROUP BY card_number
 HAVING COUNT(*) > 3;"

TIME STREAMING

TUMBLING HOPPING SESSION

リアルタイム監視
イベント（IoT、センサーなど）から洞察を導き、それを行動に変える
"CREATE TABLE failing_vehicles AS
 SELECT vehicle, COUNT(*)
 FROM vehicle_monitoring_stream
 WINDOW TUMBLING (SIZE 1 MINUTE)
 WHERE event_type = 'ERROR’
 GROUP BY vehicle
 HAVING COUNT(*) >= 3;"

データ変換
カフカの既存データを素早く派生させる
"CREATE STREAM clicks_by_user_id
 WITH (PARTITIONS=6,
             TIMESTAMP='view_time’
             VALUE_FORMAT='JSON') AS --- データをJSONに変換する
 SELECT * FROM clickstream              --- データの再入力
 PARTITION BY user_id;"

ストリームからストリームへのジョイン
例：すべてのSHIPMENTS行を2時間以内のORDERS行と照合して後期注文を検出します。
"CREATE STREAM late_orders AS
 SELECT o.orderid, o.itemid
 FROM orders o 
 FULL OUTER JOIN shipments s WITHIN 2 HOURS
 ON s.orderid = o.orderid WHERE s.orderid IS NULL;"

StreamsのINSERT INTOステートメント
  例：オンラインおよびオフラインの店舗で商品1日あたりの売上を計算する
"CREATE STREAM sales_online (itemId BIGINT, price INTEGER, shipmentId BIGINT) WITH (...);
 CREATE STREAM sales_offline (itemId BIGINT, price INTEGER, storeId BIGINT) WITH (...);
 CREATE STREAM all_sales (itemId BIGINT, price INTEGER) WITH (...); 
-- ストリームを `all_sales`にマージします。
 INSERT INTO all_sales SELECT itemId, price FROM sales_online;
 INSERT INTO all_sales SELECT itemId, price FROM sales_offline;
 CREATE TABLE daily_sales_per_item AS
 SELECT itemId, SUM(price) FROM all_sales
 WINDOW TUMBLING (SIZE 1 DAY) GROUP BY itemId;

Demo

IoTセンサー解析のための深い学習
解析モデルを使用したKSQL UDF
→KSQL文で一度書く
"SELECT event_id
 anomaly(SENSORINPUT)  ---  User Defined Function
 FROM health_sensor;"

ユーザー定義関数（UDF）

KSQLをプロダクションにする

KSQLのデプロイ

CLI REST CODE

フォールトトレランス、Kafkaから供給
分散ストリーム処理の重要な課題はフォールトトレラントな状態です。
サーバーA：
「ステートフルなストリーム処理を行います（テーブル、結合、集計など）。
     Aのローカル状態の「ストリーミングバックアップ」
サーバーの障害発生時に状態が自動的に移行される
     Aのローカル状態の「ストリーミング復元」
      Bの変更履歴トピック
サーバーB：
「状態を復元し、サーバーAが停止した場所で処理を続行します。

フォールトトレランス、Kafkaから供給
処理はデータの損失や誤操作なしに自動的にフェイルオーバーされます。
＃3は死んで＃1と＃2が引き継ぐ
   1 Kafkaコンシューマーグループの再バランスが引き起こされる
   2 ＃3の処理と状態は、Kafka経由で残りのサーバー＃1 +＃2に移行されます
＃3は戻っているので作業は再び分割されます
   3 Kafkaコンシューマーグループの再バランスが引き起こされる
   4 処理の一部。 状態はKafka経由で＃1 +＃2から＃3に移行されます

Elasticity とスケーラビリティ、Kafkaによって供給
ライブ操作中にKSQLクラスタ内のサーバを追加、削除、再起動することができます。
「より多くの処理能力が必要です！」
   1 Kafkaコンシューマーグループの再バランスが引き起こされる
   2 処理の一部。 状態はKafka経由で追加のサーバープロセスに移行されます
「うん、もう一度スケールダウンすることができます。」
   3 Kafkaコンシューマーグループの再バランスが引き起こされる
   4 処理を含む処理 停止したサーバーの状態がKafka経由で残りのサーバーに移行される

PARALLELI

KSQLはApache Kafka用のストリーミングSQLエンジンです

リソースと次のステップ
- GitHubでデモを試してください:)
- コードをチェックする
- Confluent オープンソースのダウンロード：
  https://www.confluent.io/download/
私たちとチャット：
  https://slackpass.io/confluentcommunity #ksql
  https://github.com/confluentinc/demo-scene

世界最高のストリーミングプラットフォーム - どこでも
DAVID PETERSON
システムエンジニア - Confluent APAC @davidseth

LINE Ads Platformの開発を支えるKafka

Hiromu Ogawa さん

LINE Ads Platformの開発を支えるKafka from LINE Corporation

www.slideshare.net

LINE Ads Platformで利用されているKafkaについての内容。

LINE Ads Platform とは運用型広告プラットフォームでLINE各種サービスへの配信がワンストップで可能というものだそうです。

www.linebiz.com

LAPとKsfkaの活用
#kafkajp pic.twitter.com/euoIMtyHe9
— kabukawa (@kabukawa) 2018年12月18日

LINE DMP
#kafkajp pic.twitter.com/W2pLLZ71td
— kabukawa (@kabukawa) 2018年12月18日

Kafksを利用する利点 #kafkajp pic.twitter.com/NhnxFuFYAi
— kabukawa (@kabukawa) 2018年12月18日

考慮点と実現方法 #kafkajp pic.twitter.com/CUJ5ZlOSVt
— kabukawa (@kabukawa) 2018年12月18日

I/O intensiveなKafka ConsumerアプリケーションのスループットをLINE Ads Platformではどのように改善したか

岡田遥来ジャコ(@ocadaruma) さん

I/O intensiveなKafka ConsumerアプリケーションのスループットをLINE Ads Platformではどのように改善したか from LINE Corporation

www.slideshare.net

LINE Ads Platformで利用されているKafkaでのスループット改善についての内容。

上記スライドで紹介されている、毎日2500億件のメッセージを持つLINEサービス用のマルチテナンシーKafkaクラスターについてのスライドはこちら。

Multi-Tenancy Kafka cluster for LINE services with 250 billion daily messages from LINE Corporation

www.slideshare.net

Yahoo! JAPANのKafka Platformで起きた障害とチューニング

TakumaTachibana さん

Yahoo! JAPANのデータパイプラインで起きた障害とチューニング - Apache Kafka Meetup Japan #5 - from Yahoo!デベロッパーネットワーク

www.slideshare.net

性能問題が発生した時の調査についての知見をお話しいただきました。

対処方法の検討と対策 #kafkajp pic.twitter.com/hZz3l5oAdT
— kabukawa (@kabukawa) 2018年12月18日

ログファイルのファイルディスクリプタが開放されない問題はGCの発生頻度によると突き止めた話 #kafkajp pic.twitter.com/f2W5lQzkEI
— kabukawa (@kabukawa) 2018年12月18日

懇親会

f:id:kabukawa:20181218211416j:plain:w700 f:id:kabukawa:20181218211434j:plain:w350 f:id:kabukawa:20181218211625j:plain:w350

たくさんの食べ物と飲み物をいただきました。
とても美味しかったし、お腹いっぱいになりました。
スポンサーのLINEさん、ありがとうございます！

LT1：Custom management apps for Kafka

kimutansk さん

Custom management apps for Kafka from 宗太郎木村

www.slideshare.net

LT2：prometheus-kafka-consumer-group-exporterを使った監視の話"

yosshi_ さん

speakerdeck.com

まとめ

Apache Kafka Meetup Japan へは特装版に続き2回目の参加でしたが、内容が実践的でとても勉強になりました。
ConfluentのDavid Peterson さんのセッションは英語でしたが、スライドを公開いただけたので、こうやって見返して復習することができました。
ありがとうございました！