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

BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再...

BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する

MonotaRO

May 23, 2024
Tweet

More Decks by MonotaRO

Other Decks in Technology

Transcript

  1. BigQueryとCloud Composerを使って大 規模バッチ処理をデータパイプラインに 再構築する 1 © 2023 MonotaRO Co., Ltd.

    All Rights Reserved. Masato Nakamura(@masahito) #bq_sushi #19@Google渋谷ストリーム 2024/05/23
  2. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 { "name":"Masato Nakamura", "SNS": {

     "X(a.k.a. Twitter)": "masahito",   "GitHub": "masahitojp" }, "loves": [ "Python", "JS", "Java/Scala", "Coffee☕ ", "大規模データ処理" ] } 2 自己紹介(self-introduction)
  3.   わたしたちについて 3 BtoB を対象に、 自ら間接資材の在庫を持ち、 自らオンラインで売るEC 企業 コールセンター、商 品

    採 用、物 流、 マーケティング、データサイエンス、 IT など多くの業務とシステムを 自社開発、自社運用している フルスタック EC カンパニー 事業紹介
  4. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 事例:MonotaROでの商品情報基盤 • 商品情報の検索用のミドルウェアの移行中 ◦ Solr

    -> Elasticsearchへ ▪ 検索 カテゴリーの記事 - MonotaRO Tech Blog • 商品情報データ作成処理も再実装して移行中 ◦ 商品情報 約2,217万 ◦ 商品属性名のバリエーション 約2.6万 9
  5. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 事例:MonotaROでの商品情報基盤 11 • 商品情報データ作成処理は(Python+BigQuery)のバッ チ処理で作っていた

    BigQuery ソース データ BigQuery 商品情報 データ データ変換 検索エンジン 商品情報 データ SQLをテーブルの依 存関係に従って順番 に実行 業務ロジック(変換) をPython+SQLで実 装 エラーが起きたら再 実行 外部システムとの連 携処理 やらないといけないことがたくさんある 😅
  6. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 事例:MonotaROでの商品情報基盤 12 • やらないといけないことを整理 ◦

    業務ロジックとそれ以外の処理をまずは分けた ワークフローエンジン BigQuery ソース データ BigQuery 商品情報 データ データ変換 検索エンジン 商品情報 データ 業務ロジック 注力したいのはここ! タスク依存関係の解決 外部システムとの 連携処理 リトライ処理
  7. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 13 • 処理の移行をするために、あらためて処理を見直す ワークフローエンジン BigQuery

    ソース データ BigQuery 商品情報 データ データ変換 検索エンジン 商品情報 データ 検索エンジンへの投入は システムを分ける 業務ロジック 事例:MonotaROでの商品情報基盤 注力したいのはここ! タスク依存関係の解決 リトライ処理 業務ロジックとは別でバッチ処理の機能として考えられる部分 外部システムとの 連携処理
  8. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 14 • 処理全体をフルスタックのワークフローエンジンに のせ、業務ロジックに集中すると決めた ワークフローエンジン

    BigQuery ソース データ BigQuery 商品情報 データ データ変換 検索エンジン 商品情報 データ タスク依存関係の解 決 外部システムとの連 携処理 ワークフローエンジンで実装する リトライ 業務ロジック 事例:MonotaROでの商品情報基盤 注力したいのはここ! 検索エンジンへの投入は システムを分ける
  9. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 15 • 業務ロジックに集中することで移行がスムーズに! タスク依存関係の解決や エラー処理はワークフローエ

    ンジンでカバー ワークフローエンジン BigQuery ソース データ BigQuery 商品情報 データ データ変換 業務ロジックを実装する ことに専念 BigQuery+SQLで実装 Elasticsearch 商品情報 データ 事例:MonotaROでの商品情報基盤
  10. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 • 「何を」行うか(データの移動と変換)<- これが大事 • 「どのように」行われるか

    ◦ (依存関係の管理、タスクのスケジューリング、エ ラーハンドリングなど) 16 [バッチ処理 から ワークフローへ]捉え方を変えることの効果
  11. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 Apache Airflow • 2014年にAirbnb社が開発したオープンソースのワークフローエンジン、2016年より Apache財団に寄贈。開発言語は

    Python で、処理もPythonで書ける。 • ETL(ELT)ツールと呼ばれることもある ◦ Extract、Transform、Loadの略で、複数のシステムからデータを抽出し、抽出 したデータを変換/加工した上でデータウェアハウス等へ渡す処理、およびそれ を支援するソフトウェアのこと 19
  12. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 Apache Airflow • 2014年にAirbnb社が開発したオープンソースのワークフローエンジン、2016年より Apache財団に寄贈。開発言語は

    Python で、処理もPythonで書ける。 • ETL(ELT)ツールと呼ばれることもある ◦ Extract、Transform、Loadの略で、複数のシステムからデータを抽出し、抽出 したデータを変換/加工した上でデータウェアハウス等へ渡す処理、およびそれ を支援するソフトウェアのこと ◦ -> 今回は商品情報データの変換処理(TL)なので用途に丁度あう! 20
  13. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 Apache Airflow • 特徴: 各クラウドサービスとの連携がプラグインとして提

    供されていて、主要な機能がすでに機能として使える状 態。外部システムとの連携を作りやすい。 ◦ Google Cloud - BigQuery, Cloud Dataflow, Workflows 21
  14. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 Apache AirflowとDAG • DAG (Directed

    acyclic graph; 有向非巡回グラフ)を利用 してワークフローを構築 ◦ 「依存のあるタスクをどの順番で実行するか」 22
  15. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 Apache AirflowとDAG • 利用しているDAGの例 #

    BigQueryに配置しているソースデータから新しいデータを作成 with DAG(dag_id='my_dag', ...) as dag: wait_table = BigQueryTableExistenceSenosr( # 中間テーブル作成完了を待つ task_id='wait_table',... ) execute = BigQueryInsertJobOperator( # 中間テーブルを参照して新たにテーブル作成   task_id='extract_and_transform_and_load',... ) wait_table >> execute ## タスクを列挙して、実行の順番を定義する 23
  16. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 Managed service利用の勧め • Apache Airflowの安定運用は難しい😅

    ◦ 構成要素を全部安定運用できる前提 ◦ ワーカーのスケールを考えるとK8sでの運用が必要 • マネージドサービスでの運用をお勧めしたい ◦ Google Cloud : Cloud Composer ◦ Amazon Managed Workflows for Apache Airflow 25
  17. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 Managed service利用の勧め • MonotaROの商品情報データ作成処理ではGoogle Cloud

    & Cloud Composer を採用しています • 採用したポイント ◦ 商品情報生成処理ではBigQueryを採用 ◦ Google Cloudの他サービスとの親和性の高さ 26
  18. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 運用を改善し続けるために気をつけていること • 保守性を上げるため、処理を再利用可能にする • Apache

    Airflowのバージョンアップ • アラートを意識してログを出力する • リリースを自動化する • 本番環境でのミスを減らす ◦ リリース自動化, 開発環境, ユニットテスト 28
  19. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 TaskGroupの勧め def check_source_and_create_table(wait_tables: List[str], sql_file)

    -> TaskGroup: with TaskGroup(group_id=group_id) as task_group: # 中間テーブル作成完了を待つ wait_tables = BigQueryTableExistenceSensor(..) # 中間テーブルを参照して新たにテーブル作成 create_table = BigQueryInsertJobOperator(...) wait_tables >> create_table return task_group with DAG() as dag .... ham = check_source_and_create_table([‘ham’], ‘./create_ham_table.sql’) spam = check_source_and_create_table([‘spam’], ‘./create_spam_table.sql’) eggs = check_source_and_create_table([‘eggs’], ‘./create_eggs_table.sql’) start >> [ham, spam, eggs] >> end 関数にまとめることで、 DAG の可読性が上がる 30
  20. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 TaskGroupの勧め with DAG() as dag

    .... ham = check_source_and_create_table([‘ham’], ‘./create_ham_table.sql’) spam = check_source_and_create_table([‘spam’], ‘./create_spam_table.sql’) eggs = check_source_and_create_table([‘eggs’], ‘./create_eggs_table.sql’) start >> [ham, spam, eggs] >> end 31 DAGも分かりやすくなり チームで保守しやすくなる 😄
  21. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 Apache Airflowのバージョンアップ • 各クラウドサービスでの新しいプロダクトを使う際には Apache

    Airflowのアップデートが必要です • Cloud Composerではサポート期限が設定されているので 定期的にアップデートしましょう。 • セキュリティのアップデート、UIの改善が行われており ますので定期的な更新をお勧めします 32
  22. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 事例紹介: ログとアラート # import Python

    logging module. import logging def _check_diff_field(**kwargs): # 差分を確認する if diff_set: # Data Comparison Warning: という文字列ログを残すことで Cloud Monitoring 側でアラート通知を行う logger.warning(f"Data Comparison Warning:[{diff_table_name.id}]:{diff_set}") return None check_diff_data = PythonOperator( task_id='check_diff_data', python_callable=_check_diff_field) 35
  23. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 本番環境でのミスを減らす-ロジックのユニットテスト導入 • データの変更が意図通りであることを確認する ◦ 「商品に対して販売価格を計算する」のようにある程度の

    粒度でクエリを分割し処理をおこなっているイメージ。各 クエリに対して動作を確認する。 ◦ -> 実行するBigQueryのクエリに対して、ユニットテスト を準備する ◦ -> クエリはGitHubで管理しているのでPR時点で自動実行 38
  24. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 • ユニットテストのフロー 1. 必要なデータを insert

    2. クエリを実行 3. 実行結果を結果テーブルから selectしてassert/error関数で検証 39 事例紹介: ロジックのユニットテスト導入
  25. BigQueryとCloud Composerを使って大規模バッチ処理をデータパイプラインに再構築する #bq_sushi #19@Google渋谷ストリーム 2024/05/23 今後の取り組み • 基盤として顧客の検索体験向上やサイトUXの高度化の施 策実行しやすくする ◦

    移行が進んだことで施策の実行を増やす • データ(DaaS)としてのユーザ体験の向上(データUX向上) ◦ 検索に利用するデータとして開発がスタートしたが、分析/推薦 など別の用途でも利用が拡大している。データの利用法などの整 備が必要なタイミング ▪ Dataplexなど検討中(特にデータリネージが欲しい) 42