Databricks Auto Loader & Delta Live TablesによるETL

Auto Loader & Delta Live TablesによるETLについて説明いたします。

Transcript

1. ©2024 Databricks Inc. — All rights reserved Databricks Auto Loader & Delta Live Tables によるETL Databricks Japan
2. クラウド ストレージの ランディングゾー ン シルバー (クレンジング 拡張) ゴールド (ビジネスレディ) Databricks SQL, BIツール ブロンズ (生データ バージョン管理) ノートブック, IDE, (Auto)ML Databricksランタイム Databricks SQLウェアハウス エンドユーザー Azure Data Lake Storage Gen2やAmazon S3 外部システム (ERP, オペレーショナルDB, メッ セージブローカーなど) 例 Kafka Connect, Event Hubs Capture, CDC, CSVダンプ Deltaフォーマットで整理され信頼できるテーブル 典型的なデータレイク連携
3. クラウド ストレージの ランディングゾー ン シルバー (クレンジング 拡張) ゴールド (ビジネスレディ) Databricks SQL, BIツール ブロンズ (生データ バージョン管理) ノートブック, IDE, (Auto)ML Databricksランタイム Databricks SQLウェアハウス エンドユーザー Azure Data Lake Storage Gen2やAmazon S3 外部システム (ERP, オペレーショナルDB, メッ セージブローカーなど) 例 Kafka Connect, Event Hubs Capture, CDC, CSVダンプ Deltaフォーマットで整理され信頼できるテーブル 典型的なデータレイク連携 このステップで何が問題に？
4. クラウド ストレージの ランディングゾー ン ブロンズ (生データ バージョン管理) このステップで何が問題に？ 外部システム (ERP, オペレーショナルDB, メッ セージブローカーなど) 例 Kafka Connect, Event Hubs Capture, CDC, CSVダンプ 詳しく見てみましょう • 間違っていくつかのファイルをスキップ → データの欠損 • 間違って以前のファイルを取り込み → 重複し、エラーを含むBIやレポート • DIYのファイル追跡 / 一覧はスケールせず、コスト効率が悪い • スキーマの変更 / 問題 → ジョブの失敗 • スキーマの変更 / 問題 → ファイルの損失、破損 (有害!) Azure Data Lake Storage Gen2やAmazon S3
5. ©2024 Databricks Inc. — All rights reserved Auto Loaderで取り込みを解決
6. Auto Loaderによるスケーラブルなexactly-onceのデータ取り込み JSON CSV ✅ ✅ AVRO PARQUET スキーマ 進化 • 新規データファイルがクラウドストレージに 到着するとインクリメンタルかつ効率的に処理 ◦ ファイル通知モードによってイベント駆動の取り込みを実 現(あなたの代わりに自動でEvent Grid / Amazon SNS + Azure Queue Storage / Amazon SQSをセットアップ) • 到着ファイルのスキーマを自動で推定、あるいは スキーマヒントで既知の情報を提示 • 自動のスキーマ進化 • レスキューデータ列 - 決してデータを 失いません ✅ ✅
7. Auto Loaderによるスケーラブルなexactly-onceのデータ取り込み • 新規データファイルがクラウドストレージに 到着するとインクリメンタルかつ効率的に処理 ◦ ファイル通知モードによってイベント駆動の取り込みを実 現(あなたの代わりに自動でEvent Grid / Amazon SNS + Azure Queue Storage / Amazon SQSをセットアップ) • 到着ファイルのスキーマを自動で推定、あるいはス キーマヒントで既知の情報を提示 • 自動のスキーマ進化 • レスキューデータ列 - 決してデータを 失いません JSON CSV ✅ ✅ AVRO PARQUET スキーマ 進化 df = spark .readStream .format("cloudFiles") .option("cloudFiles.format", "json") .load("abfss://…" or "s3://") .<apply your transformations> .writeStream .option("checkpointLocation","/chk/path") .start("/out/path") ✅ ✅
8. データソース / トピック バケット、コンテナ、 ディレクトリで分離 ブロンズテーブル Group 1 外部システム (ERP, オペレーショナルDB, メッ セージブローカーなど) 例 Kafka Connect, Event Hubs Capture, CDC, CSVダンプ (スキーマの異なる)数百のデータソース / トピックがある場合には？ ブロンズテーブル Group 2 ブロンズテーブル Group 3 … Databricks ジョブ 1 k 個のAuto Loader ストリーム Databricks ジョブ 2 m 個のAuto Loader ストリーム Databricks ジョブ 3 n 個のAuto Loader ストリーム ストリームはブロックしないので、いくつかのストリーム(最大 ~20)をジョブにグルーピングすることができます (これはいくつかのアプローチのうちの 一つに過ぎません) …
9. ©2024 Databricks Inc. — All rights reserved 次のステップ データエンジニアリングとオーケ ストレーションの 民主化
10. Databricksによる生産性の最大化とリスクの最小化 Auto Loader, Delta Live Tablesなど 連続あるいは スケジュールによ るデータ 取り込み 宣言型ETL パイプライン データ品質検証お よび モニタリング データ パイプラインの観 測可能性 オートスケーリン グおよび 耐障害性 自動デプロイ オペレーション パイプライン& ワークフローの オーケストレー ション チェンジデータ キャプチャー
11. • 「どのように」を抽象化し「何を」解くのかを定義す る、意図に基づく宣言型開発を使用 • 高品質なリネージを自動で作成し、データ パイプラインのテーブルの依存関係を管理 • エラー、依存関係の欠如、文法エラーを自動で チェックし、パイプラインのリカバリーを管理 • (Python UDFを含む)PythonとSQLをサポート ◦ カスタムライブラリやモデルと連携 /* アカウントテーブルの一時ビューを作成 */ CREATE INCREMENTAL LIVE VIEW account_raw AS SELECT * FROM cloud_files(“/data”, “csv”); /* ステージ 1: ブロンズテーブルで不適切な行を削除 */ CREATE INCREMENTAL LIVE TABLE account_bronze AS COMMENT "Bronze table with valid account ids" SELECT * FROM account_raw ... /* ステージ 2:シルバーに行を送信し、妥当性チェックを適用 */ CREATE INCREMENTAL LIVE TABLE account_silver AS COMMENT "Silver Accounts table with validation checks" SELECT * FROM account_bronze ... ブロンズ シルバー ゴールド ソース セルフサービス: Delta Live Tablesによる宣言型ETLパイプライン
12. ブロンズ シルバー CDC経由 UPSERT CDC経由 UPSERT CDC経由 UPSERT ストリーミング ソース クラウドオブ ジェクトストレー ジ 構造化データ 非構造化 データ 準構造化 データ データ移行サー ビス データソー ス • DBR、クラウドストレージ、DBFSでサポートされて いる任意のデータの行レベルの変更をキャプ チャー • よりシンプルなアーキテクチャ: シンプルかつイン クリメンタルなパイプラインの構築 • 順序が守られていないイベントのハンドリング • スキーマ進化 • シンプルかつ宣言型の「APPLY CHANGES INTO」APIを用いた変更レコード(insert、 update、delete)のインクリメンタルな処理 Delta Live Tablesによるチェンジデータキャプチャー(CDC)
13. Delta Live Tablesによるデータ品質検証およびモニタリング ▪ データエクスペクテーションでパイプラインにおけ るデータ品質、完全性を定義 ▪ 柔軟なポリシーによるデータ品質エラーへの対応 (失敗、削除、警告、検疫) ▪ 全てのデータパイプライン実行と品質メトリクスを 記録、追跡、レポート /* ステージ 1: ブロンズテーブルで不正な行を削除 */ CREATE INCREMENTAL LIVE TABLE fire_account_bronze AS ( CONSTRAINT valid_account_open_dt EXPECT (acconut_dt is not null and (account_close_dt > account_open_dt)) ON VIOLATION DROP ROW COMMENT "適切なアカウント IDを含むブロンズテーブル " SELECT * FROM fire_account_raw ...
14. Delta Live Tablesによるデータ品質検証およびモニタリング • データエクスペクテーションでパイプラインにおけ るデータ品質、完全性を定義 • 柔軟なポリシーによるデータ品質エラーへの対 応(失敗、削除、警告、検疫) • 行レベルのオペレーション、ガバナンス、 品質、ステータスのきめ細かいロギング /* ステージ 1: ブロンズテーブルで不正な行を削除 */ CREATE INCREMENTAL LIVE TABLE fire_account_bronze AS ( CONSTRAINT valid_account_open_dt EXPECT (acconut_dt is not null and (account_close_dt > account_open_dt)) ON VIOLATION DROP ROW COMMENT "適切なアカウント IDを含むブロンズテーブル " SELECT * FROM fire_account_raw ...
15. • バックログ認知のスケーリング意思決定によるストリー ミングSLOの達成 - スケールアップ、スケールダウンす るために、バックログメトリクスとクラスター利用率の両 方をモニタリング • 自動エラーハンドリングと容易な際実行によるダウンタ イムの削減 • すべてのDelta Liveテーブルの自動最適化によるメンテ ナンスの排除 • ジョブを並列化しデータ移動を最小化する弾力性のある Apache Spark™ベースの計算クラスターを自動で配備 し、データパイプラインワークロードを処理 Streaming source Spark executors No/Small backlog & low utilization Backlog monitoring Utilization monitoring Scale down Delta Live Tablesによるオートスケーリング、耐障害性
16. まとめ • Auto Loaderによってデータ取り込みを 堅牢かつスケーラブルに • Delta Live Tables はエンドツー エンドでデータパイプラインを管理し、可 視性を提供 • 基盤としてのDelta Lakeが、データの バージョン管理、信頼性、パフォーマンス を充当
17. Additional Resources • Auto Loader • Getting Started with Delta Live Tables • 5 Steps to Implementing Intelligent Data Pipelines With Delta Live Tables • Product Page • Documentation