Database Inside Out - Apache Kafka®️ と ksqlDB®️ によって広がるデータ活用

BigData-JAWS 勉強会#20 - 登壇資料

加速度的に広がるデータのサイズや種類に対して、様々なデータストアやデータ基盤を活用して、これまで不可能だった体験や新たな価値を提供する。この無理ゲーに対して、我々はより大きなデータストアを求め、より高い並列処理能力を駆使する挑戦を続けています。

本セッションでは少し異なる観点 - 分散するデータをメッシュとして繋ぐ、Apache KafkaとksqlDBによるDatabase Inside Outの概念についてお話しします。

Transcript

1. BigData-JAWS 勉強会 #20 Database Inside Out - with Apache Kafka® and ksqlDB® Shinichi Hashitani, Solutions Engineer, Apr. 2022. #bdJAWS
2. Apache Kafka on AWS #bdJAWS
3. 3
 Challenges 常に同時接続3万人以上のユーザーがゲームにログイン。 既存システムでは100msのレイテンシーがボトルネックと なり、ゲーム内にバッファリングが発生しエンドユーザー 体験が著しく低下。フェアなプレイ環境が維持できずユー ザー離れが進む。 Solutions Kafkaによって30ms以下の低遅延の機能を提供する事で ゲームアクションの可視性を高め、エンドユーザーの体験 を向上。合わせてバトルフィールドの健全性を正確に把握 する為より多くのメトリクスを集積/解析。チート行為の検 出の精度を向上し、合わせてより素早いアクションを自動 的に取れるよう機能を追加。 Real Time Online FPS Game
4. 4
 Data MeshにおけるData Productとしてス トリームを各サービスが提供 - 各サービス が内部でトランザクションデータを利用形 態に合わせて内部で変換。Data Productは セルフサービスで各サービス/分析チーム が直接利用。 データはKafkaを流れ、内部での処理なら びに利用はTopicとして保持される。 Souce/Sinkとの接続はKafka Connect、 データ変換にはApache Flink、スキーマ管 理はConfluent Schema Registryの上に独 自のスキーマデータモデルのソリューショ ンを利用。 Flink Forward 2020 Streaming Data Mesh by Netflix
5. @ShinHashitani | developer.confluent.io 全ての決済トランザクションでApache Kafkaを利用。イベントをデータのSouce of Truthとして捉えるEvent Storeの概念 を採用。 Kafkaの全てのログを恒久的に保存するア プローチを模索。Tiered Storageを活用 することにより、決済トランザクションの 全履歴を格納/利用することが可能に。 監査目的だけではなく、保全されたイベン トをデータ解析等にも利用を広げる。 Payment Transactions
6. Apache Kafka® #bdJAWS
7. @ShinHashitani | developer.confluent.io Microservice Challenges - Orchestration Independent Scalability コンポーネント単位でデプロイ&ス ケール可能。アプリケーションの 成長に応じて個々が独自のライフ サイクルによってサービスを管 理。個々のスケールが可能 - ボト ルネックとなる局所のみの増強に より全体スループットを向上。 Cascading Failure (連鎖障害) あるサービスが応答出来なくなると、その サービスにリクエストするサービスが応答不 能となる。1サービスの障害、高負荷による反 応速度低下が全体的な停止/パフォーマンス低 下に繋がる。 Data Consistency サービス毎に独立したデータストアを持つ - 異なるサービス間でデータ整合性を保つ必要 がある。2PCやSagaの様なパターンではデー タ同期コストが高く処理も複雑になる。ス ケールしにくい。
8. @ShinHashitani | developer.confluent.io Kafka-Based Durable Event Driven Architecture Pull Model Pushモデルである為バックプレッシャーが不要 - Consumerをスケールする事により、他に影響を与え ず個別に対応。また一般的なMessage Brokerとは異 なり、同じストリームのConsumerが増えてもBroker の負荷にほとんど影響を与えない。 Durable Streams Brokerが任意の期間ストリームを保持できる為、 Consumeによる遅延は発生してもデータの欠損を起 こさない。Consumerが追いつく為の時間的バッ ファーが生まれる。
9. @ShinHashitani | developer.confluent.io CQRS and Change Data Capture CQRS Command Query Responsibility Segregation - CRUD (Command) と検索 (Query) の役割とデータストアを分離するア プローチ。既存システムへの影響を抑えつ つ、ボトルネックになりがちな検索機能を別 のサービスとして切り離す。 Change Data Capture データベースの更新を抽出/イベント化し連携 することで、別データストア間のデータ整合 性を非同期に保つ手法。更新イベントは漏れ なく、順序通り連携する必要がある。 Command Query
10. @ShinHashitani | developer.confluent.io Stream Processing - Data Processing in Transit Event Store 分散ストレージというアーキテクチャを採用 - イベントは自動的にレプリケートされ保全対象 となる。保全期間は数時間から数ヶ月、用途 (ストリーム) 単位に柔軟に設定が可能。 イベントは保全期間内であれば何度でもリプレ イ可能。 Event Streams 個々のサービスはKafkaからイベントを受領、 加工し、アウトプットとしてイベントを出力。 アプリケーションは非同期処理のチェーンに よって構成され、処理の流れがストリームとし て組織全体を流れる。
11. @ShinHashitani | developer.confluent.io final Properties settings = new Properties(); settings.put(ProducerConfig.CLIENT_ID_CONFIG, driverId); settings.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092"); settings.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); settings.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class); settings.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://schema-registry:8081"); final KafkaProducer<String, PositionValue> producer = new KafkaProducer<>(settings); ... final ProducerRecord<String, PositionValue> record = new ProducerRecord<>(“store-order”, key, value); producer.send(record); どこにどう送るか？ 何を送るか？ How
12. @ShinHashitani | developer.confluent.io final ProducerRecord<String, PositionValue> record = new ProducerRecord<>(“store-order”, key, value); 何のイベントか？ イベント自体 “store-order” is a Topic Kafkaにイベントを送る上で指定する唯一の「何のイベントか」に関わる情報。 KafkaはこのTopic毎にイベントをまとめて保存する。
13. @ShinHashitani | developer.confluent.io Stream = Grouped Events customer login: abc order confirmed: #001 order updated: #002 customer login: efg order canceled: #003 package received: #a01 at dist center: #b02 left dist center: #a02 delivered: #a01 customer C: 0001 order U: 0003 payment U: 0002 payment C: 0003 customer U: 0002 store-order order confirmed: #001 order updated: #002 order canceled: #003 store-customer customer login: abc customer login: efg logistic package received: #a01 left dist center: #a02 delivered: #a01 at dist center: #b02 orderdb-c customer C: 0001 customer U: 0002 orderdb-o order U: 0003 orderdb-p payment C: 0003 payment U: 0002
14. @ShinHashitani | developer.confluent.io Why Kafka? - Kafka Keeps Data Consistent 14 イベントはトランザクションログとして保存 イベントはログとして永続化され、同じイベン トを何度でも読み込み処理する事が可能。Pullモ デルでもある為、イベントを漏れなく順序通り 高速に連携出来る仕組みとなっている。 customer login order confirmed order updated customer logout order canceled Append-Only Immutable 1 2 3 4 5 6 8 7 10 9 11 12 1 2 3 4 5 6 8 7 Old New
15. @ShinHashitani | developer.confluent.io Event, Total Order, and Data Consistency 15 “Streams and Tables in Apache Kafka: A Primer”, Michael Noll, Confluent Blog. チェスの一手一手とチェス盤の状態は 同じデータの異なる表現方法。 • チェス盤はある特定時点での完全な状 態 (State) を表現できる。 • チェスの一手一手を漏れなく、順序通 り適用すればチェス盤の状態を再現で きる。
16. @ShinHashitani | developer.confluent.io Log • Continuous • Logical • Infinite • Continuous • Physical • Finite Stream
17. @ShinHashitani | developer.confluent.io Kafka is a Durable Storage Broker 1 Broker 2 Broker 3 Broker 4 Topic1 partition1 Topic1 partition2 Topic1 partition3 Topic1 partition4 Topic1 partition1 Topic1 partition1 Topic1 partition2 Topic1 partition2 Topic1 partition3 Topic1 partition3 Topic1 partition4 Topic1 partition4 Concurrent Access Data Replication
18. Hello, ksqlDB! #bdJAWS
19. @ShinHashitani | developer.confluent.io Stream Processing with Kafka DB CONNECTOR CONNECTOR APP APP DB STREAM PROCESSING CONNECTOR APP DB
20. @ShinHashitani | developer.confluent.io Connecting to Kafka - Data In/Out DB CONNECTOR CONNECTOR APP APP DB STREAM PROCESSING CONNECTOR APP DB
21. @ShinHashitani | developer.confluent.io Stream Processing - Handling Data DB CONNECTOR CONNECTOR APP APP DB STREAM PROCESSING CONNECTOR APP DB
22. @ShinHashitani | developer.confluent.io ksqlDB Abstraction DB APP APP DB PULL PUSH CONNECTORS STREAM PROCESSING MATERIALIZED VIEWS ksqlDB APP
23. @yourtwitterhandle | developer.confluent.io Easily Build Real-Time Apps Joins Aggregates Push & Pull Queries Filters User-Defined Functions Connectors Compute Storage ksqlDB Kafka CREATE TABLE activePromotions AS SELECT rideId, qualifyPromotion(distanceToDst) AS promotion FROM locations GROUP BY rideId EMIT CHANGES 複雑なリアルタイム処理アプリを 数行のSQLで表現 ksqlDB at a glance ksqlDB is a database for building real-time applications that leverage stream processing
24. @ShinHashitani | developer.confluent.io Event and State - Pull Query and Push Query Seven Eleven payment: 500 JPY CREATE STREAM payments AS SELECT Account_id, store, amount FROM transactions EMIT CHANGES; CREATE TABLE payment_sum AS SELECT account_id, SUM (amount) FROM transactions GROUP BY account_id TUMBLING WINDOW (1 DAY) EMIT CHANGES; Yoshinoya payment: 800 JPY AEON payment: 2,400 JPY hashi: 3700 JPY PULL hashi: Seven Eleven, 500 JPY hashi: Yoshinoya, 800 JPY hashi: AEON, 2400 JPY PUSH hashi: 500 JPY hashi: 1300 JPY hashi: 3700 JPY PUSH
25. ksqlDB in Action
26. @ShinHashitani | developer.confluent.io Analyzing Site Access stream table process pageviews KSQL Female View Site users Admin female_per_region female_views KSQL Per Region KSQL Top Viewer KSQL Region 8&9 female_top_viewer female_region_89 CREATE STREAM female_views AS SELECT USERS.ID USERID, PAGEVIEWS.PAGEID PAGEID, USERS.REGIONID REGIONID, USERS.GENDER GENDER FROM pageviews LEFT OUTER JOIN USERS users ON ((pageviews.USERID = users.ID)) WHERE (users.GENDER = 'FEMALE') EMIT CHANGES; CREATE TABLE female_per_region AS SELECT views.GENDER GENDER, views.REGIONID REGIONID, COUNT(*) NUMUSERS FROM FEMALE_VIEWS views WINDOW TUMBLING ( SIZE 30 SECONDS ) GROUP BY views.GENDER, views.REGIONID HAVING (COUNT(*) > 1) ; CREATE STREAM female_top_viewer AS SELECT * FROM female_views views WHERE (CAST(SPLIT(views.PAGEID, '_')[2] AS INTEGER) >= 50) EMIT CHANGES; CREATE STREAM female_region_89 AS SELECT * FROM female_views views WHERE ((views.REGIONID LIKE '%_8') OR (views.REGIONID LIKE '%_9')) EMIT CHANGES; リアルタイムアクセス解析 • サイトから2種類のストリーム - ユーザー情報はログイン時にイベ ント化。JOIN対象ユーザーを 絞って解析。 • 一次処理した結果ストリームを異 なるダッシュボード毎に処理。 • ニーズによってストリームとテー ブルの混在。
27. @ShinHashitani | developer.confluent.io ksqlDB - JSON Struct { "id": "STBCKS289803838HHDHD", "transaction": { "num_shares": 50000, "amount": 50044568.89, "txn_ts": "2020-11-18 02:31:43", "customer": { "first_name": "Jill", "last_name": "Smith", "id": 1234567, "email": "jsmith@gmail.com" }, "company": { "name": "ACME Corp", "ticker": "ACMC", "id": "ACME837275222752952", "address": "Anytown USA, 333333" } } } CREATE STREAM TRANSACTION_STREAM ( id VARCHAR, trans STRUCT< num_shares INT, amount DOUBLE, txn_ts VARCHAR, customer STRUCT< first_name VARCHAR, last_name VARCHAR, id INT, email VARCHAR>, company STRUCT< name VARCHAR, ticker VARCHAR, id VARCHAR, address VARCHAR> >); SELECT TRANS->num_shares AS SHARES, TRANS->CUSTOMER->ID AS CUST_ID, TRANS->COMPANY->TICKER AS SYMBOL FROM TRANSACTION_STREAM EMIT CHANGES; SHARES CUST_ID SYMBOL 5000 1234567 ACMC 3000 2345678 IMPC 20000 3456789 HECH
28. @ShinHashitani | developer.confluent.io ksqlDB - Functions CAST CAST(COL0 AS BIGINT) GREATEST GREATEST(col1, col2, …) LEAST LEAST(col1, col2, …) LN LN(col1) ROUND ROUND(col1, scale) ARRAY_CONCAT ARRAY_CONCAT(array1, array2) ARRAY_DISTINCT ARRAY_DISTINCT([1, 1, 2, 2, 3]) ARRAY_MAX ARRAY_MAX(['foo', 'bar', 'baz']) ARRAY_SORT ARRAY_SORT(['foo', 'bar', 'baz'], 'ASC|DESC') Scalar Functionの一覧 JSON_ARRAY_CONTAI NS JSON_ARRAY_CONTAINS('[1, 2, 3]', 3) SLICE SLICE(col1, from, to) CONCAT CONCAT(col1, col2, 'hello', …, col-n) INSTR INSTR(string, substring, [position], [occurrence]) ENCODE ENCODE(col1, input_encoding, output_encoding) EXTRACTJSONFIELD EXTRACTJSONFIELD(message, '$.log.cloud') REGEXP_EXTRACT REGEXP_EXTRACT('(([AEIOU]).)', col1, 2)
29. Database Inside Out #bdJAWS
30. @ShinHashitani | developer.confluent.io Connecting Data Stores APP APP
31. @ShinHashitani | developer.confluent.io Kafka and ksqlDB - Database Inside Out fsync buffer load 1 2 3 4 5 6 8 7 10 9 11 12 1 2 3 4 5 6 8 7 Tables Materialized Views Query Engine Storage
32. Your Apache Kafka® journey begins here developer.confluent.io