Databricks Certified Data Engineer Professional 試験勉強まとめ

試験勉強に使用した教材

公式

• Data Engineer Professional Exam Guide
• Platform Architecture
• クラシック コンピュート プレーン ネットワーキング
• アクセス制御リスト
• ETL のテーブルとビュー
• GDPR コンプライアンスのためのデータの準備
• Databricks によるデータエンジニアリング
• 予測的最適化 for Unity Catalog マネージドテーブル
• コンピュート構成の推奨事項
• Databricks でテーブルをパーティション分割する場合
• リキッドクラスタリングを使用して Delta テーブル
• Delta Lake を構成してデータ ファイルのサイズを制御する
• merge を使用した Delta Lake テーブルへのアップサート
• Delta table ストリーミング 読み取りと書き込み
• 構造化ストリーミングの出力モードを選択する
• Databricks で構造化ストリーミングのバッチサイズを構成する
• Delta Lakeで チェンジデータフィードを使用する
• foreachBatch を使用して任意のデータシンクに書き込む
• パイプラインの期待値によるデータ品質の管理
• Taking Apache Spark’s Structured Streaming to Production
• Apache Spark Structured Streaming Programming Guide
• public Structured Streaming API
• Databricks REST API reference
• ジョブ API 2.0
• Spark UI Simulator
• Web UI

Udemy

• Databricks Certified Data Engineer Professional - Preparation
• Practice Exams: Databricks Certified Data Engineer Professional

その他

• Databricks Certified Data Engineer Professional Exam
• How Delta Lake Liquid Clustering conceptually works

試験結果

• Tooling : 83 %
• Data Processing : 72 %
• Data Modeling : 91 %
• Security : 83 %
• Monitoring : 100 %
• Testing and Deployment : 83 %

試験対策

知識

Databricks アーキテクチャ

Platform Architecture

• コントロールプレーンとサーバレスコンピュートプレーン間の接続は、パブリックインターネットではなく、常にクラウドネットワークバックボーン経由
• コントロールプレーンと従来のコンピュートプレーン間の接続
  • セキュアなクラスター接続
    • 顧客 VPC にはオープンポートがなく、クラシックコンピュートプレーンリソースにはパブリック IP アドレスがない
    • クラスター・ノード上のパブリックIPアドレスを必要とせずに、HTTPS(ポート443)を使用したセキュア・トンネルを介してクラスターがコントロール・プレーンに接続
  • VPC Peering（VPCピアリング）
    • Databricksが提供するコントロールプレーンと、ユーザーのクラウド環境のVPCをピアリングして通信を確立
  • プライベート接続の有効化
    • AWS PrivateLinkを使用して、クラシックコンピュートプレーンからコントロールプレーンへのプライベート接続を確立

ドライバーノードのターミナル表示

• View タブ → Web Terminal からドライバーノードのターミナルを表示可能

Delta Lake

特徴

• マネージドテーブル、外部テーブルともに適用可能（マネージドテーブルは Delta Lake 形式のみサポート）
• ACIDトランザクションを保証
• バッチ・ストリーミング処理を統一されたインターフェースで提供
• データファイルをParquetとして、トランザクションログをJsonとして保持
• トランザクションログを使用して、テーブルに対する変更を確認し、以前のテーブルバージョンをクエリ可能
• 書き込み時にスキーマを検証し、テーブルに書き込まれたすべてのデータが設定した要件と一致することを確認
• メタデータ解析と物理データレイアウトを組み合わせて、ファイルIOの数を減らし、処理の高速化を実現
• 10コミットごとに、それまでのトランザクションの内容が集約されたチェックポイントファイルが作成され、作成後はこのファイルとそれ以降のJSONログを読み込むだけで済むため、ログの読み取りコストが低減し、高速なデータアクセスが可能になる。さらにログファイルの効率的な管理にもつながる。
  • チェックポイントファイルの中身を確認するコード

    display(spark.read.parquet(f"{save_path}/_delta_log/00000000000000000010.checkpoint.parquet"))
    

• Deltaテーブルの特定の機能を削除することも可能
  • コード

    ALTER TABLE <table-name> DROP FEATURE <feature-name>
    

  • 削除可能な機能
    • checkConstraints
    • collations-preview
    • columnMapping
    • deletionVectors
    • typeWidening-preview
    • v2Checkpoint

Liquid Clustering

• 特徴
  • 高速で柔軟
  • 過剰パーティショニングや過少パーティショニングに陥ることなく、適切なパーティショニングを自律的に実現
  • 新規データに対しても、逐次的に部分クラスタリングを実施
  • 一貫したファイルサイズを維持し、データのスキューに頑健
  • 並行処理により適している
  • ストリームテーブルやマテリアライズドビューを含む、全てのDeltaテーブルで適用可能
  • パーティション分割や ZORDERと互換性がない
• 恩恵のあるシナリオ
  • カーディナリティの高い列によってフィルタリングされることが多いテーブル
  • データの分散に大きな偏りがあるテーブル
  • 急速に増大し、メンテナンスとチューニングが必要となるテーブル
  • 並列書き込み要件のあるテーブル
  • 時間の経過と共に変化するアクセスパターンを持つテーブル
  • 一般的なパーティションキーでは、パーティションが多すぎたり少なすぎたりするテーブル
• クエリ
  • 新規作成：CREATE OR REPLACE TABLE new_table CLUSTER BY col AS (SELECT * FROM source_table)
  • 既存テーブルへの適用：ALTER TABLE table CLUSTER BY col

Deletion Vectors

• 特徴
  • 従来のDelta Lakeでは、行を削除・更新する際に、該当するデータファイル全体を書き換える必要があったが、Deletion Vectorsを用いることで、削除・更新された行を論理的にマークするだけで、物理的なデータファイルの書き換えを遅延させることが可能となり、処理が高速になる
  • 削除・更新・結合（Merge）はDeletion Vector File (削除または更新された行の情報を保持する別ファイル) に記録
  • 物理的なデータファイルの書き換えは、OPTIMIZEコマンド実行時・オートコンパクション実行時・REORG TABLE ... APPLY (PURGE)実行時に発生
• 恩恵
  • 大規模なテーブルでの削除および更新操作が高速化
  • データファイルの書き換え回数が減るため、ストレージコストを削減
  • 削除・更新操作と同時に、他のクエリを実行できるようになり、コンカレンシーが向上
• 注意点
  • 削除ベクトルを有効にすると、テーブルプロトコルがアップグレードされ、削除ベクトルをサポートしていない Delta Lake クライアントでは、テーブルを読み取ることができなくなる
  • マテリアライズドビューまたはストリーミングテーブルを作成するときは手動で有効にする必要がある (ALTERステートメントは使用できず、CREATE時に有効化が必要 (CREATE TABLE <table-name> [options] TBLPROPERTIES ('delta.enableDeletionVectors' = true);))

CDF (チェンジデータフィード)

• Deltaテーブルのバージョン間の行レベルの変更を追跡
• CDFを有効化する方法
  • 新規テーブル：CREATE TABLE student (id INT, name STRING) TBLPROPERTIES (delta.enableChangeDataFeed = true)
  • 既存テーブル：ALTER TABLE myDeltaTable SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
• チェンジデータフィードを読み取る場合は、オプション readChangeFeed を true に設定 (.option("readChangeFeed", "true"))
• ターゲットテーブルに、特定の時点 (N) までに適切な変更が加えられたすべてのレコードがすでに含まれている場合は、次の時点 (N＋1) の開始バージョンを指定 (.option("startingVersion", N＋1))
• チェンジデータフィードには、Deltaテーブルのスキーマのデータ列に加えて、変更イベントの種類を識別するメタデータ列が含まれる
  • _change_type：insert, update_preimage , update_postimage, delete (1)
  • _commit_version：変更を含むDeltaログまたはテーブルのバージョン
  • _commit_timestamp：コミットが作成されたときに関連付けられたタイムスタンプ
• 変更データはテーブル ディレクトリの下の _change_data フォルダーに記録される

列マッピング

• 特徴
  • メタデータのみの変更で、データ ファイルを書き換えることなく列を削除または名前変更を実施可能
  • Parquet で許可されていない文字 (スペースなど) を使用して Delta テーブルの列に名前を付けることができるため、文字制約による列名の変更を行わずとも、CSV または JSON データを Delta に直接取り込み可能となる
• コード
  • 列名変更：ALTER TABLE <table-name> RENAME COLUMN old_col_name TO new_col_name
  • 列の削除：ALTER TABLE table_name DROP COLUMNS (col_name_1, col_name_2, ...)

生成列

ベストプラクティス

• テーブル削除をディレクトリ全体を削除として実行するのではなく、REPLACE TABLEによる上書きを実行する
  • ディレクトリを再帰的にリストアップしたり、ファイルを削除したりする必要がないため、はるかに高速
  • 古いバージョンのテーブルが存在するため、タイムトラベル可能
  • アトミックなオペレーションであり、同時実行クエリーは、テーブルを削除している間もテーブルを読み取ることが可能
  • ディレクトリ全体を削除するよりも、Vacuumを使用して古いファイルを削除するほうが高速
• データ破損やデータ損失の問題を回避するために、Databricks では、S3 に格納されている同じ Delta テーブルを異なるワークスペースから変更しない

パフォーマンス最適化

• キャッシュ
  • ディスクキャッシュ
    • Databricks独自の機能 (Delta Lakeとの関係なし)
    • ファイルをリモートのロケーションからフェッチする必要がある場合、データは自動的にキャッシュ
    • ワーカー ノードで提供されるローカル SSD で使用可能なスペースの最大半分を使用するように構成される
  • Apache Sparkキャッシュ
    • Sparkで頻繁にアクセスするデータフレームをキャッシュする機能
  • 比較表

    機能	ディスクキャッシュ	Apache Sparkキャッシュ
    保存形式	ワーカーノード上のローカルファイルとして保存	メモリ内ブロック
    適用対象	S3、ABFS、およびその他のファイルシステムに保存されているすべての Parquetテーブル	任意のDataFrameまたはRDD
    トリガー	最初の読み取り時に自動的に実行（キャッシュが有効な場合）	コードの変更が必要
    評価	遅延評価	遅延評価
    可用性	構成フラグを使用して有効または無効にすることができ、特定のノードタイプではデフォルトで有効	いつでも利用可能
    排除	LRU方式またはファイル変更時に自動で、クラスターの再起動時に手動で排除される	LRU方式では自動的に、unpersist では手動で排除される

• 動的ファイルプルーニング
  • フィルタ文やWHERE句を含むクエリに対して適用される
  • パーティション化されていないテーブルや、パーティション化されていないカラムのジョインに特に効率的
  • 有効な条件
    • Photonが使用されていない時：SELECT のみ有効
    • Photonが使用されている時：MERGE, UPDATE, DELETE でも有効
• Predictive I/O
  • データのスキャンと読み取りにかかる時間を短縮
  • 更新、削除、およびマージ中に書き換える必要のあるデータの量を削減
  • Databricks の Photon エンジン専用の機能
  • ディープラーニング技術を活用
• CBO（コストベースオプティマイザ）
  • テーブルと列の統計を収集し、適切なクエリ順序に組み換え
• AQE (アダプティブクエリ実行)
  • クエリの実行中に発生するクエリの再最適化
  • シャッフルおよびブロードキャスト終了時の最新の統計に従って、クエリを再最適化

マネージドテーブルと外部テーブルの比較

Predictive Optimization

• OPTIMIZE、VACUUM、ANALYZEオペレーションのメリットがあるテーブルを特定し、自動でオペレーションを行う機能
• Liquid Clustering を使用したテーブルに対して、Optimize/Vacuumを自動化
• 管理対象テーブルにデータが書き込まれるときに統計を収集
• すべての操作がシステムテーブル (system.storage.predictive_optimization_operations_history)にログ記録される

CLONE

• DEEP CLONE：テーブルのメタデータとデータファイルの両方を完全にコピー
• SHALLOW CLONE：テーブルのメタデータのみをコピーし、データファイル自体はソーステーブルを参照

データ構造のアクセス

• : : JSON文字列のフィールドにアクセス (schema_of_json() コマンドでJSON文字列のスキーマを取得)
• . : 構造体のフィールドにアクセス (from_json() コマンドでJSON文字列を構造体に変換)
• SELECT json.* FROM (SELECT from_json(json_string) as json) でJSON文字列のキーを列名、値を各行の要素として展開

ワークフロー

• Jobs (1つ以上のタスクから構成)
  • ノートブック
  • Pythonスクリプト
  • Python Wheel
  • SQL
  • ダッシュボード
  • DLT
  • dbt
  • JARファイル
  • Sparkサブミット
• フロー
  • 逐次
  • 並列
  • 条件分岐
  • モジュール
• トリガー
  • マニュアル
  • Cron
  • API
  • ファイル到着
  • テーブル更新
  • ストリーミング
• コンピュート
  • 汎用クラスタ
  • ジョブクラスタ
  • サーバレス
• 作成方法
  • UI
  • CLI
  • Terraform
  • API
• Tips
  • クラスタのメモリサイズを大きくしすぎると、GCのコストが増加するので、適切なメモリサイズを選択
  • 適切なメモリが設定されている場合、処理はネットワークバウンドもしくはCPUバウンドになる
  • CPU使用率は常に80%を超えるようにコア数を選択する（目安は8-16）
• パラメータ

  質問	Job Parameters	Job Contexts	Task Values
  説明	ジョブ内の各タスクに渡されるパラメータ。パラメータのフォーマットや挙動はタスクタイプに依存。	特別なテンプレート変数のセットであり、ジョブやタスク実行に関する内向きのメタデータを提供。	カスタムパラメータ。Databricksジョブ内のタスク間で共有可能。
  設定方法	各タスクのUI上 または Jobs API へのJSON定義で設定	サポートされている変数はジョブパラメータとして渡す	Databricks TaskValues サブユーティリティ
  取得方法	タスクタイプに依存	Databricks Widgets API	Databricks TaskValues サブユーティリティ
  使用理由	タスクタイプに依存	Run IDやジョブの開始時間など、ジョブ実行に関するコンテキストを渡すため	Databricksジョブ内でタスク間の情報を共有するため

Delta Live Tables と SQL/Python+ジョブ の比較

Databricks Asset Bundles (DAB)

• アーティファクト、ソースコード、構成ファイルなどをまとめて管理
• 開発環境、テスト環境、本番環境など、異なる環境間での一貫性を保ちながらデプロイ可能
• 主なコマンド
  • バンドルの検証：databricks bundle validate
  • デプロイ：databricks bundle deploy
  • ジョブの実行：databricks bundle run <job-name>
• DLTのYAMLは VIEW settings YAML から確認可能
• JobのYAMLは Switch to code version（YAML）から確認可能

環境の分割方法

クエリライフサイクル（Unity Catalog Security Model）

• クエリの送信
  • ユーザー（Principal）が SELECT * FROM table1; を送信
  • Unity Catalogに対応したクラスターまたはSQLウェアハウス）にクエリが届く
• メタデータのチェック
  • クラスタはUnity Catalog（メタデータ管理を行うControl Plane）に問い合わせを行い、**テーブルのメタデータ、ビュー、SQL関数、アクセスコントロールリスト（ACL）**を確認
• IAMロールの取得
  • Unity Catalog は、適切なIAMロールまたはサービスプリンシパルを取得し、アクセス権限を確認
• 一時的な認証情報の発行
  • Unity Catalog は、短期間有効なトークンと署名付きURLを発行
• データ取得リクエスト
  • クエリ実行のため、クラスタは署名付きURLを使用してクラウドストレージにデータをリクエスト
• データ取得
  • クラウドストレージ（Data Plane）からデータが返される
• クラスタポリシーの適用
  • クエリ実行時にクラスタ側でポリシーが適用され、データアクセスが制御される
• クエリ結果の返却
  • ユーザーに最終的なクエリ結果が返される

Unity Catalog におけるデータアクセスの権限継承モデル

• 権限は、カタログ、スキーマ、テーブルの階層構造に従って継承される
• 明示的な権限付与は、継承された権限を上書き可能
• 権限は、ユーザー、グループ、サービスプリンシパルに対して付与できる

アクセス制御

• ワークスペース管理者：ワークスペース内のすべてのオブジェクトに対する CAN MANAGE 権限を持つ
• ユーザー：作成したオブジェクトに対する CAN MANAGE 権限を自動的に持つ

Access Control Lists (ACLs)

テーブルへの制約付加

• ALTER TABLE table ADD CONSTRAINT col CHECK condition;
• DESCRIBE EXTENDED で制約の確認可能

行ごと・列ごとのアクセス制御

• 行フィルタリングUDF

  CREATE OR REPLACE FUNCTION filter_rows(column_value INT)
  RETURNS BOOLEAN
  RETURN column_value > 10;
  

• カラムマスキングUDF

  CREATE OR REPLACE FUNCTION mask_column(column_value STRING)
  RETURNS STRING
  RETURN IF(CURRENT_USER() = 'admin', column_value, '***');
  

• テーブルの作成とUDFの適用

  CREATE OR REPLACE TABLE my_table (
      id INT,
      name STRING,
      value INT
  )
  USING DELTA
  LOCATION '/path/to/my_table'
  ROW FILTER filter_rows(value)
  COLUMN MASK name USING mask_column(name);
  

• 既存テーブルへのUDFの適用

  ALTER TABLE existing_table
  SET ROW FILTER filter_rows(value)
  ALTER COLUMN name SET MASK USING mask_column(name);
  

• UDFの適用状況の確認

  DESCRIBE EXTENDED my_table;
  

Databricks カタログタイプ

Lakehouse Federation

• 統一された権限制御とインテリジェントなプッシュダウン最適化により、すべてのデータへの安全なアクセスを提供する単一のポイントを提供
• データがどこに存在していても、データの検出、クエリ、管理が可能
• 統一された権限制御により、すべてのデータソースに対するアクセス管理を一元化
• インテリジェントなプッシュダウン最適化により、クエリパフォーマンスを向上
• 幅広いデータソースをサポートしており、既存のデータインフラストラクチャとの統合が容易
• 外部データは Catalog Explorer の External data から連携可能
  • カタログ作成時に、Foreign タイプを選択
  • Connection type から使用する外部データの種類を選択

Marketplace

• Delta Sharingを利用
• データセット、機械学習（ML）モデル、ノートブック、アプリケーション、ダッシュボードなど、さまざまなデータ資産を共有・取得する場を提供
• クラウド、リージョン、プラットフォームを超えたオープンでシームレスな共有

Delta Sharing

• テーブル、ビュー、ファイル、モデルなどを共有するための機能
• オープンプロトコルを使用して、クロスプラットフォームでデータを共有可能
• レプリケーションなしでデータを共有可能であり、二重管理のリスクを排除でき、ストレージコストを削減可能
• Delta Sharingサーバーによりデータプロバイダーが管理するデータとAI資産へのアクセスを制御
• データコンシューマーは様々なツールやプログラミング言語を使用してデータにアクセス可能

パフォーマンス

• 推奨事項

  領域	推奨事項	詳細
  自動チューニング	Databricksと Delta Lakeの自動チューニングを活用	ファイルサイズの自動調整、自動最適化による小ファイル問題の回避、AQEによる自動スキュー処理
  データスキッピング	Z-Orderを活用	カーディナリティの高いカラムでZ-Orderインデックスを作成
  統計情報の収集	テーブル統計情報の収集	特に結合に使用される列の統計情報を収集
  パーティショニング	低カーディナリティ列でのデータスキッピング	年、月、日などの低カーディナリティ列でパーティショニングを使用（1TBを超えるテーブルに対してのみ適用）
  CDC アーキテクチャ	SQL DML機能の活用	変更データのみを処理するCDCアーキテクチャへの移行
  リソース管理	ジョブクラスターとSQLウェアハウスの分離	リソース競合を回避するためにジョブクラスターとSQLウェアハウスを分離
  コード最適化	本番ジョブでのアクション操作を回避	読み書き以外のアクションをトリガーする操作を避ける（count(), display(), collect()など）
  コード最適化	ドライバーノードへの計算集中を回避	シングルスレッドのPython/Pandasの代わりにPandas API on Sparkを使用し、関数を分散処理
  コード最適化	Python UDFの行単位実行を回避	行ごとのPython UDFの代わりにネイティブのPySpark関数またはベクトル化されたPandas UDFを使用
  コード最適化	最適化されたデータ構造の使用	RDDの代わりにDataFramesまたはDatasetsを使用し、コストベースのオプティマイザを活用（RDDはコストベースのオプティマイザを活用できない）

• ボトルネック

  症状	詳細	対策
  スキュー	処理対象データサイズの不均衡により、一部のノードに負荷が集中	AQE｜スキューが発生しているファイルのフィルタ
  スピル	メモリ不足による一時ファイルのディスクへの書き込みが発生し、パフォーマンスが低下	クラスタのコア当たりのRAMを増やす｜スキューを解消しデータサイズの不均衡を抑制｜パーティションサイズの管理｜explode() のような高コストな操作を避ける
  シャッフル	ワイドトランスフォーメーションを行う際に、ワーカーノード間でデータを移動することで、ネットワーク負荷が増大	少数のサイズの大きいワーカーを使用し、ネットワークIOを低減する｜NVMeやSSDを使用し、IO速度を高速化する｜不必要な行や列をフィルタリングし、データシャッフルの量を減らす
  スモールファイル	小さなファイルが大量にあると管理のオーバーヘッドが増加し、処理効率が悪化	Liquid Clustering｜スキューの解消
  シリアライズ	データやコードの変換・転送時に処理遅延が発生	組み込み関数を使う（UDFを使わない）｜（UDFを使う必要がある場合）ベクトル化UDFを使用

• SQLウェアハウスのパフォーマンス改善

  • Total wall-clock duration
    • Scheduling
      • Waiting for compute の時間が長い場合
        • サーバレス or 起動済みクラスタを増やす
      • Waiting for queue の時間が長い場合
        • クラスタの最大数を増やす
    • Running
      • Optimization query & pruning files の時間が長い場合
        • 統計情報とファイル最適化が必要
      • Executing の時間が長い場合
        • プランの改善もしくはコンピューティングリソースの追加が必要

• クエリ最適化

  • Unresolved Logical Plan
    • シンタックスや文法のエラーを確認
  • Logical Plan
    • メタデータ（カラムのデータ型など）のカタログやランタイムの統計情報を活用して、論理的に適切なクエリ計画を作成
  • Optimized Logical Plan
    • Catalyst Catalogを活用して計画を最適化
  • Physical Plans
    • クラスタやコンピュート内で行う実際のステージやタスクを計画
  • Selected Physical Plan
    • Cost Based Optimizationにより選択されたPhysical Plansの実行

• 分散データ処理におけるJoin戦略

  • クロスJOINは非常にコストがかかるため、低レイテンシーまたは頻繁な再計算を必要とするワークロードとクエリからクロスJOINは削除
  • 複数のJOINを実行する場合は、必ず最初に最小のテーブルをJOINし、次に大きなテーブルとJOIN
  • 予測的最適化を有効にし、最新の統計を保持
    • 2024年11月11日以降に作成されたすべてのアカウントにてデフォルトで有効
    • ストリーミングテーブルとマテリアライズドビューでは有効化できない

  Join戦略	説明	最適な使用シナリオ	特徴
  Reordering the join	ジョイン順序を最適化することで実行効率を向上させる手法	複数のテーブルを結合する複雑なクエリ	テーブルサイズや選択率に基づいて最適な順序を決定｜中間結果のサイズを小さくしてメモリ使用量を削減｜クエリオプティマイザが自動で実行することが多い
  Dynamically Switching Join Strategies	実行時のデータ特性に基づいて最適なジョイン方法（Broadcast / Shuffle / Sort）を動的に選択する手法	データサイズや分布が予測困難な場合	実行時の統計情報に基づいて戦略を選択｜システムリソースに応じて適応｜パフォーマンスの自動最適化
  Broadcast Hash Join	小さいテーブルを全ノードにブロードキャスト（複製）し、大きいテーブルとハッシュベースで結合	一方のテーブルが十分小さく、メモリに収まる場合	シャッフルのオーバーヘッドを削減｜小テーブル-大テーブルの結合に最適｜ネットワーク転送コストが発生
  Shuffle Hash Joins（Databricks Photonのデフォルト）	ハッシュ化した結合キーに基づいて両方のテーブルのデータをパーティション化し、同じハッシュ値を持つレコードが同じノードに集めて結合	大きなテーブル同士の結合	データを均等に分散｜各ノードでハッシュテーブルを構築｜スケーラビリティが高いがシャッフルコストが発生
  Sort-Merge Join（OS Sparkのデフォルト）	両方のテーブルを結合キーでソートし、ソートされたデータをマージするように走査	メモリが限られた環境での大規模データセット結合	メモリ効率が良い｜ソートのコストが発生するが結合処理は効率的｜データサイズに対するスケーラビリティが高い

クラスタ最適化のベストプラクティス

• スポットインスタンスをワーカーノードで有効にする
• 可能な限り最新のLTS Databricks Runtimeを使用する
• 適用可能な場合はPhotonを使用し、TCOを最適化する
• 最新世代のVMを使用し、汎用VMから始め、メモリ/計算最適化VMをテストする

データ変換の種類

• 宣言型
  • 目的の結果をどのように達成するかではなく、望ましい結果に焦点を当てる
  • 可読性が高い
  • メンテナンスコストが低い
• 手続き型
  • 明示的な命令による計算の実行、データを操作するための操作の正確なシーケンスを定義
  • 柔軟性が高く、複雑な処理を実装可能
  • メンテナンスコストが高い

バッチ処理とストリーミング処理

• マテリアライズドビューは、通常バッチ処理の保証に従うが、可能な場合はDLTを使用して結果を増分的に計算する
• ステートフル集計を使用してデータセット全体の統計を計算することはできず、データセット全体の増分集計計算を行う場合は Materialized View の使用が推奨されている
• ステートフルな構造化ストリーミングクエリの最適化のための推奨事項
  • コンピュート最適化インスタンスをワーカーとして使用
  • シャッフル パーティションの数を、クラスターのコア数の 1 倍から 2 倍に設定
  • spark.sql.streaming.noDataMicroBatches.enabled を false に設定（データを含まないマイクロバッチを処理しないように設定）

結合

• バッチ結合
  • ステートレス
• Stream-Stream 結合
  • ステートフル
  • データソースと結果に関する情報を追跡し、結果を繰り返し更新
  • 結合の両側にウォーターマークを定義する必要がある
  • 状態データを自動的にチェックポイントし、再起動後に同じスキーマとして復元するため、再起動間で以下のようなスキーマを変更してはならない
    • ストリーミング集計でのグループ化キー・集計の数・タイプの変更
    • ストリーミング重複排除でのグループ化キー・集計の数・タイプの変更
    • ストリーム-ストリーム join でのスキーマの変更やジョインの種類 (外部または内部) の変更
  • サポートされているJOINタイプ
    • Inner join
    • Left outer join
    • Right outer join
    • Full outer join
    • Left semi join
  • コード

    import dlt
    
    dlt.create_streaming_table("adImpressionClicks")
    @dlt.append_flow(target = "adImpressionClicks")
    def joinClicksAndImpressions():
      clicksDf = (read_stream("rawClicks")
        .withWatermark("clickTimestamp", "3 minutes")
      )
      impressionsDf = (read_stream("rawAdImpressions")
        .withWatermark("impressionTimestamp", "3 minutes")
      )
      joinDf = impressionsDf.alias("imp").join(
      clicksDf.alias("click"),
      expr("""
        imp.userId = click.userId AND
        clickAdId = impressionAdId AND
        clickTimestamp >= impressionTimestamp AND
        clickTimestamp <= impressionTimestamp + interval 3 minutes
      """),
      "inner"
      ).select("imp.userId", "impressionAdId", "clickTimestamp", "impressionSeconds")
    
      return joinDf
    

    from pyspark.sql import functions as F
    from pyspark.sql.window import Window
    
    def batch_upsert(microBatchDF, batchId):
        window = Window.partitionBy("order_id", "customer_id").orderBy(F.col("_commit_timestamp").desc())
    
        (microBatchDF.filter(F.col("_change_type").isin(["insert", "update_postimage"]))
                    .withColumn("rank", F.rank().over(window))
                    .filter("rank = 1")
                    .drop("rank", "_change_type", "_commit_version")
                    .withColumnRenamed("_commit_timestamp", "processed_timestamp")
                    .createOrReplaceTempView("ranked_updates"))
    
        query = """
            MERGE INTO customers_orders c
            USING ranked_updates r
            ON c.order_id=r.order_id AND c.customer_id=r.customer_id
                WHEN MATCHED AND c.processed_timestamp < r.processed_timestamp
                  THEN UPDATE SET *
                WHEN NOT MATCHED
                  THEN INSERT *
        """
    
        microBatchDF.sparkSession.sql(query)
    
    def process_customers_orders():
        orders_df = spark.readStream.table("orders_silver")
    
        cdf_customers_df = (spark.readStream
                                .option("readChangeData", True)
                                .option("startingVersion", 2)
                                .table("customers_silver")
                          )
    
        query = (orders_df
                    .join(cdf_customers_df, ["customer_id"], "inner")
                    .writeStream
                        .foreachBatch(batch_upsert)
                        .option("checkpointLocation", "dbfs:/mnt/demo_pro/checkpoints/customers_orders")
                        .trigger(availableNow=True)
                        .start()
                )
    
        query.awaitTermination()
    
    process_customers_orders()
    

• Stream-Static 結合
  • ステートレス
  • 増分データソース (ファクトテーブルなど) と静的データソース (緩やかに変化するディメンションテーブルなど) を結合するための適切なオプションを提供
  • ストリーミング ソースから新しく受信したレコードのみが現在のバージョンの静的テーブルと結合される
    • ストリームテーブルへのデータ追加のみでトリガーされ、静的テーブルへのデータ追加ではトリガーされない
    • クエリがトリガーされた際は、常に最新の静的テーブルが参照される
  • コード

    streamingDF = spark.readStream.table("orders")
    staticDF = spark.read.table("customers")
    
    query = (streamingDF
      .join(staticDF, streamingDF.customer_id==staticDF.id, "inner")
      .writeStream
      .option("checkpointLocation", checkpoint_path)
      .table("orders_with_customer_info")
    )
    

    from pyspark.sql import functions as F
    
    def process_books_sales():
    
        orders_df = (spark.readStream.table("orders_silver")
                            .withColumn("book", F.explode("books"))
                    )
    
        books_df = spark.read.table("current_books")
    
        query = (orders_df
                      .join(books_df, orders_df.book.book_id == books_df.book_id, "inner")
                      .writeStream
                        .outputMode("append")
                        .option("checkpointLocation", "dbfs:/mnt/demo_pro/checkpoints/books_sales")
                        .trigger(availableNow=True)
                        .table("books_sales")
        )
    
        query.awaitTermination()
    
    process_books_sales()
    

構造化ストリーミング

トリガータイプ

アウトプットモード

• ステートレスストリーミングの場合、すべての出力モードは同じように動作（常にそのトリガー中に処理されたレコードを出力）
• ステートフルストリーミングの場合、出力モードごとに異なる動作

  出力モード	説明
  Append (デフォルト)	ウォーターマークを考慮の上、将来のトリガーで変更されない行のみを出力 (確定した情報を取得したい場合に有効)
  Update	出力されたレコードが後続のトリガーで変更される可能性がある場合でも、トリガー中に変更されたすべての行を出力 (集計結果をリアルタイムに追跡したい場合に有効)
  Complete (ストリーミング集計でのみ機能)	生成されたすべての結果の行をダウンストリームに出力

ウィンドウタイプ

ウォーターマーク

• 遅延データ（late data）を管理するための時間的しきい値
• 古いステート情報を自動的に削除し、メモリ使用量を最適化
• ウィンドウベースの集計で効果的に機能
• スクリプト

  # イベント時間列とウォーターマークの遅延閾値を指定
  streaming_df = streaming_df.withWatermark("eventTime", "10 minutes")
  
  # ウォーターマーク内に到着した重複レコードの削除
  streaming_df = streaming_df.withWatermark("eventTime", "10 minutes").dropDuplicatesWithinWatermark(["id"])
  

Slowly Changing Dimension

• Type-2 SCDの処理コード

  • 処理関数

    def type2_upsert(microBatchDF, batch):
        microBatchDF.createOrReplaceTempView("updates")
    
        sql_query = """
            MERGE INTO books_silver
            USING (
                SELECT updates.book_id as merge_key, updates.*
                FROM updates
    
                UNION ALL
    
                SELECT NULL as merge_key, updates.*
                FROM updates
                JOIN books_silver ON updates.book_id = books_silver.book_id
                WHERE books_silver.current = true AND updates.price <> books_silver.price
              ) staged_updates
            ON books_silver.book_id = merge_key
            WHEN MATCHED AND books_silver.current = true AND books_silver.price <> staged_updates.price THEN
              UPDATE SET current = false, end_date = staged_updates.updated
            WHEN NOT MATCHED THEN
            INSERT (book_id, title, author, price, current, effective_date, end_date)
              VALUES (staged_updates.book_id, staged_updates.title, staged_updates.author, staged_updates.price, true, staged_updates.updated, NULL)
        """
    
        microBatchDF.sparkSession.sql(sql_query)
    

  • 処理実行

    def process_books():
        schema = "book_id STRING, title STRING, author STRING, price DOUBLE, updated TIMESTAMP"
    
        query = (spark.readStream
                        .table("bronze")
                        .filter("topic = 'books'")
                        .select(F.from_json(F.col("value").cast("string"), schema).alias("v"))
                        .select("v.*")
                    .writeStream
                        .foreachBatch(type2_upsert)
                        .option("checkpointLocation", "dbfs:/mnt/demo_pro/checkpoints/books_silver")
                        .trigger(availableNow=True)
                        .start()
                )
    
        query.awaitTermination()
    
    process_books()
    

• DLTでは通常、CDC（Change Data Capture）やインクリメンタル処理を活用して効率的に更新

• Full Refresh Allを指定すると、すべてのデータで再計算を実施

• 構造化ストリーミングクエリの入力レートを制限するため、一貫したバッチ サイズを維持し、大規模なバッチによってスピルや連鎖的なマイクロバッチ処理の遅延が発生するのを防ぐことが可能

  • maxFilesPerTrigger (デフォルト 1000)：各マイクロバッチで処理されるファイル数の上限
  • maxBytesPerTrigger (デフォルト 1000)：各マイクロバッチで処理されるデータ量のおおよその最大値

MERGE INTOとAPPLY CHANGES INTOの違い

foreachBatch

• ストリーミングクエリのすべてのマイクロバッチの出力データにバッチ関数を適用
• foreachBatch で使用される関数は、以下 2 つのパラメーターを受け取る
  • マイクロバッチの出力データの DataFrame
  • マイクロバッチの ID
• ストリーミングクエリの出力を複数の場所に書き込む必要がある場合、foreachBatch を使用して複数のシンクに書き込むと、ストリーミング書き込みの実行がシリアル化され、各マイクロバッチの待機時間が長くなる可能性があるため、最適な並列化とスループットのために、複数の構造化ストリーミングライターの使用が推奨される

CDCのコード

• Static

  customers_df = (spark.table("bronze")
                  .filter("topic = 'customers'")
                  .select(F.from_json(F.col("value").cast("string"), schema).alias("v"))
                  .select("v.*")
                  .filter(F.col("row_status").isin(["insert", "update"])))
  
  from pyspark.sql.window import Window
  
  window = Window.partitionBy("customer_id").orderBy(F.col("row_time").desc())
  
  ranked_df = (customers_df.withColumn("rank", F.rank().over(window))
                            .filter("rank == 1")
                            .drop("rank"))
  

• Stream

  • 間違い

    • ストリーミング DataFrame では、時間ベースの順序付けを含んでいるかに関わらず、非集計ウィンドウ関数（F.rank()、F.row_number() など）はサポートされていない

    # ストリーミングDataFramesでは非時間ベースのウィンドウ操作はサポートされていないため, 例外が発生する
    ranked_df = (spark.readStream
                      .table("bronze")
                      .filter("topic = 'customers'")
                      .select(F.from_json(F.col("value").cast("string"), schema).alias("v"))
                      .select("v.*")
                      .filter(F.col("row_status").isin(["insert", "update"]))
                      .withColumn("rank", F.rank().over(window))
                      .filter("rank == 1")
                      .drop("rank")
                )
    

  • 正解

    from pyspark.sql.window import Window
    
    def batch_upsert(microBatchDF, batchId):
        window = Window.partitionBy("customer_id").orderBy(F.col("row_time").desc())
    
        (microBatchDF.filter(F.col("row_status").isin(["insert", "update"]))
                    .withColumn("rank", F.rank().over(window))
                    .filter("rank == 1")
                    .drop("rank")
                    .createOrReplaceTempView("ranked_updates"))
    
        query = """
            MERGE INTO customers_silver c
            USING ranked_updates r
            ON c.customer_id=r.customer_id
                WHEN MATCHED AND c.row_time < r.row_time
                  THEN UPDATE SET *
                WHEN NOT MATCHED
                  THEN INSERT *
        """
    
    microBatchDF.sparkSession.sql(query)
    
    query = (spark.readStream
                      .table("bronze")
                      .filter("topic = 'customers'")
                      .select(F.from_json(F.col("value").cast("string"), schema).alias("v"))
                      .select("v.*")
                      .join(F.broadcast(df_country_lookup), F.col("country_code") == F.col("code") , "inner")
                  .writeStream
                      .foreachBatch(batch_upsert)
                      .option("checkpointLocation", "dbfs:/mnt/demo_pro/checkpoints/customers_silver")
                      .trigger(availableNow=True)
                      .start()
              )
    
    query.awaitTermination()
    

• 本番運用を想定した一連の流れ

  catalog = "jobs"
  schema = "myschema"
  employees_cdf_table = "employees_cdf"
  employees_table = "employees_merge"
  
  # チェンジデータフィードをストリームとして読み取る（各マイクロバッチに対して、MERGE INTOコマンドを実行）
  def upsertToDelta(microBatchDF, batchId):
    # マイクロバッチで各キーの最大 sequenceNum を持つ行のすべての列を取得（マイクロバッチ内の同じキーの中で最新の処理を反映）
    microBatchDF = microBatchDF.groupBy("id").agg(
      max_by(struct("*"), "sequenceNum").alias("row")
    ).select("row.*").createOrReplaceTempView("updates")
  
    # 以下の処理を実行
    # １．新しいデータのタイムスタンプが、格納されているタイムスタンプよりも大きいかどうかを確認し、大きい場合はターゲットからの新しいデータを使用、それ以外の場合は元のデータを保持
    # ２．変更データレコードが削除を示しているときは該当行を削除（実際に削除すると、その後アップサートの行が来た時に対応できないため、削除を記録）
    spark.sql(f"""
      MERGE INTO {catalog}.{schema}.{employees_table} t
      USING updates s
      ON s.id = t.id
      WHEN MATCHED AND s.operation = 'DELETE' THEN UPDATE SET DELETED_AT=now()
      WHEN MATCHED THEN UPDATE SET
        name=CASE WHEN s.sequenceNum > t.sequenceNum THEN s.name ELSE t.name END,
        age=CASE WHEN s.sequenceNum > t.sequenceNum THEN s.age ELSE t.age END,
        country=CASE WHEN s.sequenceNum > t.sequenceNum THEN s.country ELSE t.country END
      WHEN NOT MATCHED THEN INSERT *
    """)
  
  def create_table():
    spark.sql(f"DROP TABLE IF EXISTS {catalog}.{schema}.{employees_table}")
    spark.sql(f"""
    CREATE TABLE IF NOT EXISTS {catalog}.{schema}.{employees_table}
    (id INT, name STRING, age INT, country STRING)
    """)
  
  create_table()
  
  cdcData = spark.readStream.table(f"{catalog}.{schema}.{employees_cdf_table}")
  
  cdcData.writeStream \
    .foreachBatch(upsertToDelta) \
    .outputMode("append") \
    .start()
  

  -- ターゲット テーブルを直接使用できないため、クエリを実行できるビューを作成
  CREATE VIEW employees_v AS
  SELECT * FROM employees_merge
  WHERE DELETED_AT = NULL
  
  -- 削除が記録された行を定期的に削除するクリーンアップジョブを作成
  DELETE FROM employees_merge
  WHERE DELETED_AT < now() - INTERVAL 1 DAY
  

その他のコード

• ソースデータの確認

  /* Discover your data in a volume */
  LIST "/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>"
  
  /* Preview your data in a volume */
  SELECT * FROM read_files("/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>") LIMIT 10
  
  /* Discover your data in an external location */
  LIST "s3://<bucket>/<path>/<folder>"
  
  /* Preview your data */
  SELECT * FROM read_files("s3://<bucket>/<path>/<folder>") LIMIT 10
  

• 新しいデータのみを取り込む（includeExistingFiles を false に設定し、テーブル作成後にディレクトリに到着したデータのみ処理）

  CREATE OR REFRESH STREAMING TABLE my_bronze_table
  AS SELECT *
  FROM STREAM read_files('s3://mybucket/analysis/*/*/*.json', includeExistingFiles => false)
  

• ストリーミング テーブルを完全に更新（データの全履歴を保持しないソースや、保持期間が短いソース ( Kafkaなど) で完全更新を呼び出すことは、既存のデータが切り捨てられるため非推奨）

  REFRESH STREAMING TABLE my_bronze_table FULL
  

• ストリーミングレコードの重複排除
  • 次の例では、データは clickTimestamp順に処理され、重複する userId 列と clickAdId 列を含むレコードが 5 秒以内に到着した場合、重複したレコードは削除される
  • コード

    clicksDedupDf = (
      spark.readStream.table
        .option("withEventTimeOrder", "true")
        .table("rawClicks")
        .withWatermark("clickTimestamp", "5 seconds")
        .dropDuplicatesWithinWatermark(["userId", "clickAdId"]))
    

  • 一連の流れ
    • Stream読込み
      • 読み込んだマイクロバッチごとに重複排除を実行

      deduped_df = (spark.readStream
                        .table("bronze")
                        .filter("topic = 'orders'")
                        .select(F.from_json(F.col("value").cast("string"), json_schema).alias("v"))
                        .select("v.*")
                        .withWatermark("order_timestamp", "30 seconds")
                        .dropDuplicates(["order_id", "order_timestamp"]))
      

    • マイクロバッチ関数の作成
      • 読み込むマイクロバッチ内での重複削除は、ストリーム読込みで実施済みだが、以前のマイクロバッチで取り込み済みのIDがある可能性があるため、新規IDのデータのみインサート

      def upsert_data(microBatchDF, batch):
          microBatchDF.createOrReplaceTempView("orders_microbatch")
      
          sql_query = """
            MERGE INTO orders_silver a
            USING orders_microbatch b
            ON a.order_id=b.order_id AND a.order_timestamp=b.order_timestamp
            WHEN NOT MATCHED THEN INSERT *
          """
      
          microBatchDF.sparkSession.sql(sql_query)
          #microBatchDF._jdf.sparkSession().sql(sql_query)
      

    • Stream書出し
      • 読み込んだマイクロバッチごとに関数を適用

      query = (deduped_df.writeStream
                        .foreachBatch(upsert_data)
                        .option("checkpointLocation", "dbfs:/mnt/demo_pro/checkpoints/orders_silver")
                        .trigger(availableNow=True)
                        .start())
      
      query.awaitTermination()
      

Kafkaからデータを取り込むようにストリーミングテーブルを構成

• Bronzeテーブル

  import dlt
  
  @dlt.table
  def kafka_raw():
    return (
      spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "<server:ip>")
        .option("subscribe", "topic1")
        .option("startingOffsets", "latest")
        .load()
    )
  

• Silverテーブル

  CREATE OR REFRESH STREAMING TABLE streaming_silver_table
  AS SELECT
    *
  FROM
    STREAM(kafka_raw)
  WHERE ...
  

本番運用の考慮事項

• display や count などの結果を返す不要なコードをノートブックから削除
• 常にジョブコンピュートを使用して、ストリームをジョブとしてスケジュール
• コンピュートの自動スケーリングを有効にしない
• Continuousモードを使用してジョブをスケジュール

RocksDBの活用

• チェックポイント管理に RocksDB を使用（各クエリには異なるチェックポイントが必要であり、複数のクエリで同じ場所を共有しない）
• RocksDB ベースの状態管理を有効にするには、SparkSession をストリーミングクエリを開始する前に以下のコマンドを実行

  spark.conf.set(
  "spark.sql.streaming.stateStore.providerClass",
  "com.databricks.sql.streaming.state.RocksDBStateStoreProvider")
  

• RocksDB 状態ストアメトリクス
  • ストリーミングデータ処理において、特に RocksDB 状態ストア（各状態演算子（state operator）が管理する、状態管理のために使用されるデータベース）を使用している場合に収集される重要なメトリクス
  • 状態管理の監視、パフォーマンスやリソースの監視と最適化、デバッグとトラブルシューティングなどのために使用される
  • メトリクスの種類
    • インスタンス固有のメトリクス：パーティションID + ストア名 でラベル付けされる特定の状態ストアインスタンスに関連した集計値
    • 集計メトリクス：ジョブ内のすべての状態演算子の動作の合計として集計され、ジョブ全体のパフォーマンス把握に有用
• 変更履歴 (changelog) のチェックポイントを有効化
  • 従来の方式との違い
    • 従来: RocksDB状態ストアのスナップショット全体をチェックポイント処理中に耐久性のあるストレージ（通常はクラウドストレージ）にアップロード
    • 新方式: 最後のチェックポイント以降に変更されたレコードのみを書き込む増分方式
  • メリット
    • チェックポイント処理の期間短縮: 変更されたデータのみを書き込むため処理が高速
    • エンドツーエンドの待機時間削減: ストリーミングクエリのレイテンシが改善
    • ストレージ効率の向上: 変更データのみを保存するため、保存容量が削減]
    • リソース使用量の削減: データ転送量とCPU/メモリ使用量が減少
  • 設定方法

    spark.conf.set("spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled", "true")
    

• 非同期チェックポイント：ジョブの状態情報を非同期で保存（デフォルトの in-memory 状態ストアの実装ではサポート対象外）
  • 状態チェックポイント処理が完了するのを待たずに、前のマイクロバッチの計算が完了するとすぐに、次のマイクロバッチの処理を開始
  • 同期チェックポイントに比べて、レイテンシの短縮が期待できる（状態チェックポイントのレイテンシは、バッチ実行の全体的なレイテンシの主な要因の 1 つ）
  • 設定方法

    spark.conf.set(
      "spark.databricks.streaming.statefulOperator.asyncCheckpoint.enabled",
      "true"
    )
    
    spark.conf.set(
      "spark.sql.streaming.stateStore.providerClass",
      "com.databricks.sql.streaming.state.RocksDBStateStoreProvider"
    )
    

• 非同期プログレス追跡：構造化ストリーミングジョブの進捗状況を非同期で追跡
  • ジョブの進捗を監視し、リアルタイムで進捗情報を収集
  • ストリーミングクエリの進行状況追跡をバックグラウンドで実行することで、メインの処理スレッドへの影響を最小限に抑える
  • 設定方法

    spark.conf.set("spark.sql.streaming.asyncProgressTracking", "true")
    
    // stream 書出し
    val query = stream.writeStream
      .format("kafka")
      .option("topic", "out")
      .option("checkpointLocation", "/tmp/checkpoint")
      .option("asyncProgressTrackingEnabled", "true")
      .option("asyncProgressTrackingCheckpointIntervalMs", 0)
      .start()
    

ストリームの監視

• writeStream コードに .queryName() を追加してストリームに一意のクエリ名を付けると、 Spark UI内のどのメトリクスがどのストリームに属しているかを区別しやすくなる
• Apache Spark のストリーミングクエリリスナーインターフェースを使用して、ストリーミングメトリクスを外部サービスにプッシュ可能

ストリーミングの制限

• Apache Spark の連続処理モード（.trigger(continuous=“X second”)）はサポート対象外

ストリームソースとしてのビュー

• ストリームできるのは Delta テーブル形式のビューのみ
• Unity Catalog へのビューの登録が必要
• ソース ビューでサポートされている操作は、SELECT, WHERE, UNION ALL のみ

ストリーミング開始時点の指定

• startingVersion もしくは startingTimestamp を指定し、ストリームの開始時点を指定可能
• startingVersion が設定されていない場合（推奨）、その時点でのテーブルの完全なスナップショットを含む使用可能な最新バージョンからストリーム開始
• 両方のオプションを同時に設定することはできない

DLT

DLTの特徴

• 外部テーブルへの書き込みや、外部ファイルストレージやデータベーステーブルを操作する条件を組み込む処理はできない
• 制約を満たさないレコードを追加できない従来のデータベースの CHECK 制約とは異なり、期待値はデータ品質要件を満たさないデータを処理するときの柔軟性を提供
• Databricks ワークスペースでは、並列パイプライン更新の上限数は 100 個
• DLT パイプラインから発行されたマテリアライズドビュー とストリーミングテーブル (Databricks SQLによって作成されたものを含む) にアクセスできるのは、 Databricks クライアントとアプリケーションのみ
• タイムトラベルはストリーミングテーブルでのみサポートされ、マテリアライズドビューではサポート対象外
• パイプライン実行には、クラスター作成許可が必要

DLTパイプラインの例

• Python
  • @dlt.table デコレーターは、関数によって返された結果に基づいてマテリアライズドビューまたはストリーミングテーブルの作成を DLT に指示

  import dlt
  from pyspark.sql.functions import col
  
  # orders ストリーミングテーブルの作成
  @dlt.table()
  @dlt.expect_or_drop("valid_date", "order_datetime IS NOT NULL AND length(order_datetime) > 0")
  def orders():
    return (spark.readStream
      .format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load("/databricks-datasets/retail-org/sales_orders")
    )
  
  # customers マテリアライズドビューの作成
  @dlt.table()
  def customers():
      return spark.read.format("csv").option("header", True).load("/databricks-datasets/retail-org/customers")
  
  # customer_orders マテリアライズドビューの作成
  @dlt.table()
  def customer_orders():
    return (spark.read.table("orders")
      .join(spark.read.table("customers"), "customer_id")
        .select("customer_id",
          "order_number",
          "state",
          col("order_datetime").cast("int").cast("timestamp").cast("date").alias("order_date"),
        )
    )
  
  # daily_orders_by_state マテリアライズドビューの作成
  @dlt.table()
  def daily_orders_by_state():
      return (spark.read.table("customer_orders")
        .groupBy("state", "order_date")
        .count().withColumnRenamed("count", "order_count")
      )
  

• SQL

  -- orders ストリーミングテーブルの作成
  CREATE OR REFRESH STREAMING TABLE orders(
    CONSTRAINT valid_date
    EXPECT (order_datetime IS NOT NULL AND length(order_datetime) > 0)
    ON VIOLATION DROP ROW
  )
  AS SELECT * FROM STREAM read_files("/databricks-datasets/retail-org/sales_orders");
  
  -- customers マテリアライズドビューの作成
  CREATE OR REFRESH MATERIALIZED VIEW customers
  AS SELECT * FROM read_files("/databricks-datasets/retail-org/customers");
  
  -- customer_orders マテリアライズドビューの作成
  CREATE OR REFRESH MATERIALIZED VIEW customer_orders
  AS SELECT
    c.customer_id,
    o.order_number,
    c.state,
    date(timestamp(int(o.order_datetime))) order_date
  FROM orders o
  INNER JOIN customers c
  ON o.customer_id = c.customer_id;
  
  -- daily_orders_by_state マテリアライズドビューの作成
  CREATE OR REFRESH MATERIALIZED VIEW daily_orders_by_state
  AS SELECT state, order_date, count(*) order_count
  FROM customer_orders
  GROUP BY state, order_date;
  

• DLT パイプラインで MLflow モデルを使用

  %pip install mlflow
  
  import dlt
  import mlflow
  from pyspark.sql.functions import struct
  
  run_id = "mlflow_run_id"
  model_name = "the_model_name_in_run"
  model_uri = f"runs:/{run_id}/{model_name}"
  loaded_model_udf = mlflow.pyfunc.spark_udf(spark, model_uri=model_uri)
  
  @dlt.table
  def model_predictions():
    return spark.read.table(<input-data>)
      .withColumn("prediction", loaded_model_udf(struct(features))))
  

DLTシンク

• DLT シンクを使用すると、 Apache Kafka や Azure Event Hubs などのイベント ストリーミング サービスや、 Unity Catalog または Hive metastoreによって管理される外部テーブルなどのターゲットに変換されたデータを書き込むことが可能（DLT シンクの機能登場以前は、DLT パイプラインで作成されたストリーミングテーブルとマテリアライズドビューは、 Databricks 管理 Delta テーブルにのみで永続化可能だった）
• 一例として、Apache Kafka トピックなどのメッセージバスからデータを読み取り、データを低レイテンシで処理して、処理されたレコードをメッセージバスに書き戻しが可能となり、クラウドストレージからの書き込みや読み取りを行わないことで、レイテンシーを短縮できる
• シンクへの書き込みに使用できるのは append_flow のみ（apply_changesなどの他のフローはサポート対象外）
• 完全更新を実行しても、シンク内の以前のコンピュート結果データはクリーンアップされない

EXPECTの使用法

• フローチャート
• 例
  • Python

    # Simple constraint
    @dlt.expect("non_negative_price", "price >= 0")
    
    # SQL functions
    @dlt.expect("valid_date", "year(transaction_date) >= 2020")
    
    # CASE statements
    @dlt.expect("valid_order_status", """
      CASE
        WHEN type = 'ORDER' THEN status IN ('PENDING', 'COMPLETED', 'CANCELLED')
        WHEN type = 'REFUND' THEN status IN ('PENDING', 'APPROVED', 'REJECTED')
        ELSE false
      END
    """)
    
    # Multiple constraints
    @dlt.expect("non_negative_price", "price >= 0")
    @dlt.expect("valid_purchase_date", "date <= current_date()")
    
    # Complex business logic
    @dlt.expect(
      "valid_subscription_dates",
      """start_date <= end_date
        AND end_date <= current_date()
        AND start_date >= '2020-01-01'"""
    )
    
    # Complex boolean logic
    @dlt.expect("valid_order_state", """
      (status = 'ACTIVE' AND balance > 0)
      OR (status = 'PENDING' AND created_date > current_date() - INTERVAL 7 DAYS)
    """)
    
    # 無効なレコードを削除
    @dlt.expect_or_drop("valid_current_page", "current_page_id IS NOT NULL AND current_page_title IS NOT NULL")
    
    # 無効なレコードで失敗
    @dlt.expect_or_fail("valid_count", "count > 0")
    
    # 複数の制約（Pythonでのみ使用可）
    valid_pages = {"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"}
    
    @dlt.table
    @dlt.expect_all(valid_pages)
    def raw_data():
      # Create a raw dataset
    
    @dlt.table
    @dlt.expect_all_or_drop(valid_pages)
    def prepared_data():
      # Create a cleaned and prepared dataset
    
    @dlt.table
    @dlt.expect_all_or_fail(valid_pages)
    def customer_facing_data():
      # Create cleaned and prepared to share the dataset
    
    # 行数の検証
    @dlt.view(
      name="count_verification",
      comment="Validates equal row counts between tables"
    )
    @dlt.expect_or_fail("no_rows_dropped", "a_count == b_count")
    def validate_row_counts():
      return spark.sql("""
        SELECT * FROM
          (SELECT COUNT(*) AS a_count FROM table_a),
          (SELECT COUNT(*) AS b_count FROM table_b)""")
    
    # 範囲ベースの検証パターン
    @dlt.view
    def stats_validation_view():
      # Calculate statistical bounds from historical data
      bounds = spark.sql("""
        SELECT
          avg(amount) - 3 * stddev(amount) as lower_bound,
          avg(amount) + 3 * stddev(amount) as upper_bound
        FROM historical_stats
        WHERE
          date >= CURRENT_DATE() - INTERVAL 30 DAYS
      """)
    
      # Join with new data and apply bounds
      return spark.read.table("new_data").crossJoin(bounds)
    
    @dlt.table
    @dlt.expect_or_drop(
      "within_statistical_range",
      "amount BETWEEN lower_bound AND upper_bound"
    )
    def validated_amounts():
      return dlt.read("stats_validation_view")
    
    # 無効なレコードを隔離
    import dlt
    from pyspark.sql.functions import expr
    
    rules = {
      "valid_pickup_zip": "(pickup_zip IS NOT NULL)",
      "valid_dropoff_zip": "(dropoff_zip IS NOT NULL)",
    }
    quarantine_rules = "NOT({0})".format(" AND ".join(rules.values()))
    
    @dlt.view
    def raw_trips_data():
      return spark.readStream.table("samples.nyctaxi.trips")
    
    @dlt.table(
      temporary=True,
      partition_cols=["is_quarantined"],
    )
    @dlt.expect_all(rules)
    def trips_data_quarantine():
      return (
        dlt.readStream("raw_trips_data").withColumn("is_quarantined", expr(quarantine_rules))
      )
    
    @dlt.view
    def valid_trips_data():
      return dlt.read("trips_data_quarantine").filter("is_quarantined=false")
    
    @dlt.view
    def invalid_trips_data():
      return dlt.read("trips_data_quarantine").filter("is_quarantined=true")
    

  • SQL

    -- Simple constraint
    CONSTRAINT non_negative_price EXPECT (price >= 0)
    
    -- SQL functions
    CONSTRAINT valid_date EXPECT (year(transaction_date) >= 2020)
    
    -- CASE statements
    CONSTRAINT valid_order_status EXPECT (
      CASE
        WHEN type = 'ORDER' THEN status IN ('PENDING', 'COMPLETED', 'CANCELLED')
        WHEN type = 'REFUND' THEN status IN ('PENDING', 'APPROVED', 'REJECTED')
        ELSE false
      END
    )
    
    -- Multiple constraints
    CONSTRAINT non_negative_price EXPECT (price >= 0)
    CONSTRAINT valid_purchase_date EXPECT (date <= current_date())
    
    -- Complex business logic
    CONSTRAINT valid_subscription_dates EXPECT (
      start_date <= end_date
      AND end_date <= current_date()
      AND start_date >= '2020-01-01'
    )
    
    -- Complex boolean logic
    CONSTRAINT valid_order_state EXPECT (
      (status = 'ACTIVE' AND balance > 0)
      OR (status = 'PENDING' AND created_date > current_date() - INTERVAL 7 DAYS)
    )
    
    -- 無効なレコードを削除
    CONSTRAINT valid_current_page EXPECT (current_page_id IS NOT NULL and current_page_title IS NOT NULL) ON VIOLATION DROP ROW
    
    -- 無効なレコードで失敗
    CONSTRAINT valid_count EXPECT (count > 0) ON VIOLATION FAIL UPDATE
    
    -- 行数の検証
    CREATE OR REFRESH MATERIALIZED VIEW count_verification(
      CONSTRAINT no_rows_dropped EXPECT (a_count == b_count)
    ) AS SELECT * FROM
      (SELECT COUNT(*) AS a_count FROM table_a),
      (SELECT COUNT(*) AS b_count FROM table_b)
    
    -- 範囲ベースの検証パターン
    CREATE OR REFRESH MATERIALIZED VIEW stats_validation_view AS
      WITH bounds AS (
        SELECT
        avg(amount) - 3 * stddev(amount) as lower_bound,
        avg(amount) + 3 * stddev(amount) as upper_bound
        FROM historical_stats
        WHERE date >= CURRENT_DATE() - INTERVAL 30 DAYS
      )
      SELECT
        new_data.*,
        bounds.*
      FROM new_data
      CROSS JOIN bounds;
    
    CREATE OR REFRESH MATERIALIZED VIEW validated_amounts (
      CONSTRAINT within_statistical_range EXPECT (amount BETWEEN lower_bound AND upper_bound)
    )
    AS SELECT * FROM stats_validation_view;
    
    -- 無効なレコードを隔離
    CREATE TEMPORARY STREAMING LIVE VIEW raw_trips_data AS
      SELECT * FROM STREAM(samples.nyctaxi.trips);
    
    CREATE OR REFRESH TEMPORARY STREAMING TABLE trips_data_quarantine(
      -- Option 1 - merge all expectations to have a single name in the pipeline event log
      CONSTRAINT quarantined_row EXPECT (pickup_zip IS NOT NULL OR dropoff_zip IS NOT NULL),
      -- Option 2 - Keep the expectations separate, resulting in multiple entries under different names
      CONSTRAINT invalid_pickup_zip EXPECT (pickup_zip IS NOT NULL),
      CONSTRAINT invalid_dropoff_zip EXPECT (dropoff_zip IS NOT NULL)
    )
    PARTITIONED BY (is_quarantined)
    AS
      SELECT
        *,
        NOT ((pickup_zip IS NOT NULL) and (dropoff_zip IS NOT NULL)) as is_quarantined
      FROM STREAM(raw_trips_data);
    
    CREATE TEMPORARY LIVE VIEW valid_trips_data AS
    SELECT * FROM trips_data_quarantine WHERE is_quarantined=FALSE;
    
    CREATE TEMPORARY LIVE VIEW invalid_trips_data AS
    SELECT * FROM trips_data_quarantine WHERE is_quarantined=TRUE;
    

• 無効なレコードに対するアクション

  操作	SQL 構文	Python 構文	結果
  warn (デフォルト)	EXPECT	dlt.expect	無効なレコードはターゲットに書き込まれ、有効なレコードと無効なレコードの数は、他のデータセットメトリクスとともに記録される
  drop	EXPECT … ON VIOLATION DROP ROW	dlt.expect_or_drop	無効なレコードは削除され、ドロップされたレコードの数は、他のデータセットメトリクスとともに記録される
  fail	EXPECT … ON VIOLATION FAIL UPDATE	dlt.expect_or_fail	無効なレコードがある場合、そのフローが失敗し、下流の処理は中断され、途中までの処理はロールバックされる。再処理するためには手動トリガーが必要。失敗したフローと関連のないパイプライン内の他フローに対する影響はない。メトリクスは記録されない。

PythonでのDLTの記述

• dlt.read() 関数と dlt.read_stream() 関数は使用可能だが、Databricks では常に spark.read.table() 関数と spark.readStream.table() 関数の使用を推奨
  • spark 関数
    • 外部ストレージ内のデータセットや他のパイプラインで定義されたデータセットなど、内部データセットと外部データセットの両方の読み取りをサポート
    • 読み取り操作に対するオプション (skipChangeCommitsなど) の指定をサポート
  • dlt 関数
    • 内部データセットの読み取りのみをサポート
    • オプションの指定は非対応

Auto Loader

• ファイルが検出されると、そのメタデータはAuto Loaderパイプラインのチェックポイントの場所にあるスケーラブルなキーバリューストア（RocksDB）に永続化され、データが1回だけ処理されることが保証される
• ファイルが追加または上書きされた場合、ファイルのどのバージョンが処理されるかを保証できないため、不変ファイルのみを取り込み、cloudFiles.allowOverwritesの設定を避けることが推奨されている
• データファイルが連続して到着するのではなく、定期的にアップロードされる場合、予想されるファイル到着時刻の後に、Trigger.AvailableNow で実行するようにスケジュールする
• ファイルシステムの監視方法
  • ディレクトリ表示モード
    • 指定されたディレクトリ内に存在する全てのファイルを一度にリストアップして、オートローダーがそれを読み込む
    • リストアップされた全ファイルを一度に読み込むため、ファイル数が多くなると処理が重くなる可能性がある
  • ファイル通知モード
    • システムがファイルの変更（追加、削除、変更など）を監視し、変更があった場合にのみ通知を受けて、オートローダーがそのファイルにアクセス
    • ファイルが変更されると、即座に通知を受けて反応するため、リアルタイムで更新を反映させやすい
    • ディレクトリ全体をスキャンする必要がなく、変更があったファイルだけを監視するため、リソースを節約できる
    • イベント駆動型であるため、システムへの負荷が軽減される

セキュアな情報管理

PIIデータの削除

• ブロンズレイヤーからPIIデータを削除
  • コード

    spark.sql("""
    MERGE INTO target
    USING (
      SELECT user_id
      FROM gdpr_control_table
    ) AS source
    ON target.user_id = source.user_id
    WHEN MATCHED THEN DELETE
    """)
    

Dynamic Views

• 列表示に対する操作

  CREATE OR REPLACE VIEW customers_vw AS
    SELECT
      customer_id,
      CASE
        WHEN is_member('admins_demo') THEN email
        ELSE 'REDACTED'
      END AS email,
      gender
    FROM customers_silver
  

• 行表示に対する操作

  CREATE OR REPLACE VIEW customers_fr_vw AS
  SELECT * FROM customers_vw
  WHERE
    CASE
      WHEN is_member('admins_demo') THEN TRUE
      ELSE country = "France" AND row_time > "2022-01-01"
    END
  

ブロンズからシルバー、ゴールドレイヤーへの変更の伝播

• シルバー、ゴールドレイヤーがマテリアライズドビューの場合
  • ソース（ブロンズ）での削除を自動的に処理
  • Databricks によって自動的に最新の状態に保たれ（サーバレス DLT パイプラインを自動的に作成し更新処理を実施、ベーステーブルでチェンジデータフィードを有効にする必要あり（ALTER TABLE table1 SET TBLPROPERTIES (delta.enableChangeDataFeed = true);））、冗長な再計算を回避可能
• シルバー、ゴールドレイヤーがストリーミングテーブルの場合
  • ストリーミングテーブルは、追加専用データのみを処理するため、データの変更を自分で処理する必要がある
  • 推奨手順：(1) ストリーミングソース内のデータを削除、(2) ストリーミングテーブルから該当データを削除、(3) 読み取ったストリーミングを skipChangeCommits (更新や削除など挿入以外の操作をスキップするフラグ) を使用して更新
  • コード

    spark.readStream.option("skipChangeCommits", true).table("source")
    

• シルバー、ゴールドレイヤーがDelta テーブルの場合
  • 削除したことを伝達する独自コードの記述が必要
  • コード

    spark.readStream.option("readChangeFeed", true).table("source")
    

ジョブ

• ジョブとタスクでのパラメーターの使用
  • ユースケース
    • コード資産に拡張可能なロジックを追加
    • 実行を条件付け
    • 複数のタスク間で共通のパラメーターを参照
    • あるタスクで生成された情報を別のタスクで使用
    • ジョブ実行でメタデータと状態情報を参照
  • パラメータの種類
    • ジョブパラメータ
    • タスクパラメータ
    • 動的値参照
    • タスク値
• タスク間でコンピュートを共有すると、起動時間に関連するレイテンシーを短縮可能
• dbutils.widget.text(key, value) でジョブのパラメータに指定した値の動的参照が可能
• ジョブのガバナンスとして、ジョブ所有権をサービスプリンシパルに割り当てることが推奨されている。これにより、所有者の離職に対して頑健かつ、権限をサービスプリンシパルで一元管理できるようになる。ワークスペース管理者はジョブの権限を管理し、必要に応じて所有権を再割り当て可能。
• ジョブをAPI実行した場合は、Run Asのデフォルトであるオーナーではなく、APIトークンに対応するユーザーがログに記録される