Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Database Inside Out - Apache Kafka®️ と ksqlDB®️...

hashi
April 14, 2022

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

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

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

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

hashi

April 14, 2022
Tweet

More Decks by hashi

Other Decks in Technology

Transcript

  1. BigData-JAWS 勉強会 #20 Database Inside Out - with Apache Kafka®

    and ksqlDB® Shinichi Hashitani, Solutions Engineer, Apr. 2022. #bdJAWS
  2. 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
  3. @ShinHashitani | developer.confluent.io 全ての決済トランザクションでApache Kafkaを利用。イベントをデータのSouce of Truthとして捉えるEvent Storeの概念 を採用。 Kafkaの全てのログを恒久的に保存するア

    プローチを模索。Tiered Storageを活用 することにより、決済トランザクションの 全履歴を格納/利用することが可能に。 監査目的だけではなく、保全されたイベン トをデータ解析等にも利用を広げる。 Payment Transactions
  4. @ShinHashitani | developer.confluent.io Microservice Challenges - Orchestration Independent Scalability コンポーネント単位でデプロイ&ス

    ケール可能。アプリケーションの 成長に応じて個々が独自のライフ サイクルによってサービスを管 理。個々のスケールが可能 - ボト ルネックとなる局所のみの増強に より全体スループットを向上。 Cascading Failure (連鎖障害) あるサービスが応答出来なくなると、その サービスにリクエストするサービスが応答不 能となる。1サービスの障害、高負荷による反 応速度低下が全体的な停止/パフォーマンス低 下に繋がる。 Data Consistency サービス毎に独立したデータストアを持つ - 異なるサービス間でデータ整合性を保つ必要 がある。2PCやSagaの様なパターンではデー タ同期コストが高く処理も複雑になる。ス ケールしにくい。
  5. @ShinHashitani | developer.confluent.io Kafka-Based Durable Event Driven Architecture Pull Model

    Pushモデルである為バックプレッシャーが不要 - Consumerをスケールする事により、他に影響を与え ず個別に対応。また一般的なMessage Brokerとは異 なり、同じストリームのConsumerが増えてもBroker の負荷にほとんど影響を与えない。 Durable Streams Brokerが任意の期間ストリームを保持できる為、 Consumeによる遅延は発生してもデータの欠損を起 こさない。Consumerが追いつく為の時間的バッ ファーが生まれる。
  6. @ShinHashitani | developer.confluent.io CQRS and Change Data Capture CQRS Command

    Query Responsibility Segregation - CRUD (Command) と検索 (Query) の役割とデータストアを分離するア プローチ。既存システムへの影響を抑えつ つ、ボトルネックになりがちな検索機能を別 のサービスとして切り離す。 Change Data Capture データベースの更新を抽出/イベント化し連携 することで、別データストア間のデータ整合 性を非同期に保つ手法。更新イベントは漏れ なく、順序通り連携する必要がある。 Command Query
  7. @ShinHashitani | developer.confluent.io Stream Processing - Data Processing in Transit

    Event Store 分散ストレージというアーキテクチャを採用 - イベントは自動的にレプリケートされ保全対象 となる。保全期間は数時間から数ヶ月、用途 (ストリーム) 単位に柔軟に設定が可能。 イベントは保全期間内であれば何度でもリプレ イ可能。 Event Streams 個々のサービスはKafkaからイベントを受領、 加工し、アウトプットとしてイベントを出力。 アプリケーションは非同期処理のチェーンに よって構成され、処理の流れがストリームとし て組織全体を流れる。
  8. @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
  9. @ShinHashitani | developer.confluent.io final ProducerRecord<String, PositionValue> record = new ProducerRecord<>(“store-order”,

    key, value); 何のイベントか? イベント自体 “store-order” is a Topic Kafkaにイベントを送る上で指定する唯一の「何のイベントか」に関わる情報。 KafkaはこのTopic毎にイベントをまとめて保存する。
  10. @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
  11. @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
  12. @ShinHashitani | developer.confluent.io Event, Total Order, and Data Consistency 15

    “Streams and Tables in Apache Kafka: A Primer”, Michael Noll, Confluent Blog. チェスの一手一手とチェス盤の状態は 同じデータの異なる表現方法。 • チェス盤はある特定時点での完全な状 態 (State) を表現できる。 • チェスの一手一手を漏れなく、順序通 り適用すればチェス盤の状態を再現で きる。
  13. @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
  14. @ShinHashitani | developer.confluent.io Connecting to Kafka - Data In/Out DB

    CONNECTOR CONNECTOR APP APP DB STREAM PROCESSING CONNECTOR APP DB
  15. @ShinHashitani | developer.confluent.io Stream Processing - Handling Data DB CONNECTOR

    CONNECTOR APP APP DB STREAM PROCESSING CONNECTOR APP DB
  16. @ShinHashitani | developer.confluent.io ksqlDB Abstraction DB APP APP DB PULL

    PUSH CONNECTORS STREAM PROCESSING MATERIALIZED VIEWS ksqlDB APP
  17. @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
  18. @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
  19. @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対象ユーザーを 絞って解析。 • 一次処理した結果ストリームを異 なるダッシュボード毎に処理。 • ニーズによってストリームとテー ブルの混在。
  20. @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": "[email protected]" }, "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
  21. @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)
  22. @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