Databricks Certified Data Engineer Professional 試験でよく問われる内容解説

はじめに

DatabricksのCertified Data Engineer Professional試験でよく問われる項目、
理解が甘かったので学習しなおした項目をまとめました。
項目ごとに簡潔に分けているので、ざっと振り返りたい場合などにぜひご活用ください
単純にDatabricksの便利機能を知りたい場合も活用できると思います

Liquid ClusteringとZ-Orderの違い

• 「ファイルの断片化」や「スキュー（偏り）」 という単語が出てきたら、Liquid Clustering が正解の可能性が高いです。
• 「1つのカラムだけでなく、複数のカラムで効率的に絞り込みたい」 という文脈であれば、両方の手法が該当しますが、最新のベストプラクティスを問われているなら Liquid Clustering です。

mergeSchema=trueはどういう時使うべき？

• 新しいカラムの追加 (Append)
  ソースデータに新しい列（例：new_feature）が追加された際、このオプションがないと「Schema mismatch」エラーで停止します。trueにすることで、既存のテーブル構造を動的に拡張します。
• Merge文での利用
  SQLの MERGE INTO 操作を行う際、ソース側にしかないカラムをターゲットに反映させたい場合に SET T.new_col = S.new_col と記述する代わりに、スキーマ進化を許可して自動でカラムを作成させます。
• 構造化ストリーミング (Structured Streaming)
  ストリーム処理中にソースのスキーマが変わる可能性がある場合、チェックポイントと組み合わせてスキーマの変更を許容するために使用します。

expectとexpect_or_dropの違い

試験対策のポイント

試験では、「データ品質を確保しつつ、パイプラインの停止は避けたい」というシナリオが出ることがあります。その場合の正解は expect_or_drop です。

一方、「不正なデータが1件でも混入することを絶対に許さない」という要件であれば expect_or_fail を選びます。

expectとexpect_or_dropの活用例

具体的なパイプライン設計の例

1. expect の活用：データ品質のモニタリング

Bronze層からデータを読み込む際、ソースシステム側の不具合を早期発見するために使用します。

• シナリオ: ユーザーIDがたまに欠損するが、システムを止めたくない。
• コードイメージ:SQL

  CONSTRAINT valid_user_id EXPECT (user_id IS NOT NULL)

• 結果: 全レコードが保存され、Databricksの「Pipeline Details」画面で「何％のデータがNULLだったか」を視覚的に把握できます。

2. expect_or_drop の活用：データクレンジング

分析用のSilver層テーブルを作成する際に、後続のMLモデルや集計を壊さないために使用します。

• シナリオ: 年齢（age）がマイナスの値や200歳を超えるような、明らかに誤ったデータを除外したい。
• コードイメージ

CONSTRAINT realistic_age EXPECT (age > 0 AND age < 150) ON VIOLATION DROP ROW

• 結果: 正常な範囲のデータだけがテーブルに残り、異常なデータは自動的にフィルタリング（除外）されます。

試験で狙われる「落とし穴」

試験問題では、以下のような判断を求められることがあります。

• 「ドロップされたレコードはどこへ行くのか？」
  • デフォルトでは、ドロップされたレコードの内容自体はターゲットテーブルには残りません。もしドロップされた中身も調査したい場合は、expect を使って「有効フラグ列」を自作するか、Quarantine（隔離）用の別テーブルへ流すロジックを組む必要があります。
• 「複数の制約がある場合」
  • 1つのテーブルに expect と expect_or_drop を混ぜて定義することも可能です。その場合、1つでも drop 条件に触れれば、そのレコードは破棄されます。

expectすると、制約違反のデータはどこに行く？

expect の場合のデータの見え方

expect を設定したテーブルをクエリすると、制約に違反したレコードも正常なレコードと混ざって表示されます。

• 確認方法: DLTパイプラインのUI上にある「Data Quality」タブで、何パーセントのレコードが違反したかの統計を確認できます。
• 用途: 「とりあえずデータは全部入れておいて、後で WHERE 句で除外して分析する」といった運用に適しています。

2. 不正なデータを「別テーブル」に送りたい場合（隔離戦略）

試験や実務で「不正データを捨てたくないが、クリーンなテーブルにも入れたくない」という要件がある場合、DLTでは2つのテーブルに分岐させるロジックを手動で書く必要があります。

• クリーン用テーブル: expect_or_drop を使用して、正しいデータのみを保持。
• 隔離（Quarantine）用テーブル: 逆に「不正な条件」を expect（またはフィルタ）で抽出し、エラー調査用に保持。

expectやexpect_or_dropのデータや詳細データってどんな感じでアクセスする？

3. イベントログでの確認

「どのレコードがダメだったか」の統計情報は、Databricksが自動で管理するイベントログ（Event Log）という特別な場所に保存されます。

-- イベントログからデータ品質メトリクスをクエリする例
SELECT
  id,
  expectations.dataset,
  expectations.name AS constraint_name,
  expectations.passed_records,
  expectations.failed_records
FROM (
  SELECT
    id,
    explode(from_json(details:flow_progress:data_quality:expectations, 'array<struct<dataset:string, name:string, passed_records:int, failed_records:int>>')) AS expectations
  FROM event_log
  WHERE event_type = 'flow_progress'
)

例：

Windowとwith句の関係性

1. WITH 句（Common Table Expression / CTE）の意味合い

WITH 句は、メインのクエリを実行する前に「仮のテーブル」を定義するようなイメージです。

• 意味合い: 「この複雑な計算結果を temp_table という名前で呼ぶことにする」という宣言。
• メリット: サブクエリ（入れ子構造）を排除できるため、上から下に流れるようにクエリを読めるようになります。

2. WINDOW 関数の意味合い

WINDOW 関数は、集計（GROUP BY）と異なり、「元の行を維持したまま」、その行に関連する周囲のデータを参照して計算します。

• 意味合い: 「現在の行の前後 3 行の平均を出したい」「部署内での自分の順位を知りたい」といった計算。
• メリット: JOIN を繰り返さなくても、行レベルのデータと集計値を横並びにできます。

3. WITH 句と WINDOW 関数を併用する具体例

実務や試験で最も多いパターンは、「WITH 句の中で WINDOW 関数を使って順位を付け、メインクエリでその順位を使ってフィルタリングする」という流れです。

SQL での記述例

「各部署で給与が高いトップ 3 人だけを抽出したい」という場合：

-- 1. WITH句でランキング付きの仮テーブルを作る
WITH ranked_employees AS (
  SELECT 
    name, 
    dept, 
    salary,
    -- WINDOW関数で部署ごとの順位を計算
    DENSE_RANK() OVER (PARTITION BY dept ORDER BY salary DESC) as rank
  FROM employees
)

-- 2. メインクエリで順位を使って絞り込む
SELECT * FROM ranked_employees 
WHERE rank <= 3;

なぜ併用が必要なのか？

SQL の実行順序のルール上、WHERE 句の中で WINDOW 関数の結果（上記の rank）を直接使うことはできません。 そのため、一度 WITH 句で「順位という列を持ったテーブル」を確定させてから、外側でフィルタリングする必要があるのです。

具体的な違い

1. SELECT で（サブクエリを使って）始める場合

一気に結果を出そうとするアプローチです。内側から外側へ読み解く必要があります。

SELECT name, dept, salary
FROM (
  -- ここで一度ランキング付きの「中間状態」を作る
  SELECT name, dept, salary,
         RANK() OVER (PARTITION BY dept ORDER BY salary DESC) as rnk
  FROM employees
) 
WHERE rnk = 1;

テーブルの推移イメージ

1. 元データ (employees): 名前、部署、給与が並んでいる。
2. サブクエリ内: 各行の横に rnk 列が計算されて付与される。
3. 最終結果: 外側の WHERE rnk = 1 によって、1位以外の行が消える。

2. WITH 句で始める場合

「まずランキング表を作る」「次にそこから抽出する」というレシピのような手順で書くアプローチです。

- STEP 1: ランキング付きの「仮想テーブル」を定義
WITH RankedTable AS ( SELECT name, dept, salary, RANK() 
OVER (PARTITION BY dept ORDER BY salary DESC) as rnk FROM employees
)
- STEP 2: 作成した仮想テーブルから1位を抽出SELECT name, dept, salary
FROM RankedTable
WHERE rnk = 1;

STEP 1: [RankedTable] の作成（メモリ上の仮想テーブル）

STEP 2: [最終出力]（WHERE rnk = 1 でフィルタ）

hash(email)って存在する？Emailをhash化して、inputに対応したoutputを作りたい場合はどんな関数が望ましい？

hashは暗号化の関数ではない

以下が候補

最大限のセキュリティ提供はOAuthとサービスプリンシパル？

その通りです。セキュリティと管理の観点から、PATではなくサービスプリンシパルの利用が推奨されます

バンドルで作ったジョブができてる場合、runコマンドで実行可能？tオプション prodを使う？

結論から言うと、databricks bundle run コマンドで実行可能です。また、ご質問の通り、ターゲット（環境）を指定するために -t（または --target）オプションを使用するのが標準的な手順です。

databricks bundle run の仕様とオプション

もし、あなたの databricks.yml に prod というターゲットと、main_job というジョブが定義されている場合、コマンドは以下のようになります。

# prod環境のジョブを実行する場合
databricks bundle run -t prod main_job`

ワークフローで途中のタスクに勝手に必要パラメータ追加されてこけた場合、どういう再実行スタイルが望ましい？

Databricksのジョブ（ワークフロー）が途中のタスクで失敗し、特に「意図しないパラメータ追加」などの構成ミスが原因だった場合、最も効率的で望ましい再実行スタイルは 「失敗したタスクからの再実行（Repair Run）」 です。

その場合の望ましい対応フロー

「パラメータ設定ミス」でコケた場合、以下の手順がベストプラクティスです。

1. 構成の修正: まず、ジョブの設定（あるいはBundlesのソースコード）から、問題となった余計なパラメータを削除・修正します。
2. Repair Runを選択:
   • Databricks UIのジョブ実行詳細画面で 「Repair Run（修復実行）」 をクリックします。
   • これにより、「修正後の設定」を使って、失敗したタスクから処理が再開されます。
3. 部分的なパラメータ上書き (必要に応じて):
   • 再実行時に、その回だけの特定のパラメータを渡して「修復」することも可能です。

なぜ「Repair Run」が望ましいのか？

Databricksのワークフローは、各タスクが成功したという状態を保持しています。

• 効率: 既に完了している上流タスク（例：大規模なデータロード）をスキップできるため、復旧までの時間が最短になります。
• べき等性: データエンジニアリングでは、二重取り込み（Double Ingestion）を防ぐために、一度成功したタスクは繰り返さないのが基本です。

タスクとジョブはどっちがでかいくくり？

「ジョブ（Job）」の方が大きなくくりです。

1つの「ジョブ」という大きな箱の中に、実行したい個別の処理である「タスク（Task）」が複数入っている、という構造になっています。

ジョブとタスクの階層構造

ジョブの実行コマンドはrun ? execute?

Databricks CLIやAPIにおいて、ジョブを動かす際の正式な動詞（コマンド）は run です。

execute というコマンドは存在しません。Databricks Asset Bundles (DABs) でも、通常の CLI でも run を使用します。

ジョブ実行に関するコマンド体系

バンドルを定義したyamlに関して、permissionとjobの階層はどんな感じになる？

resources:
  jobs:
    my_analysis_job:  # ジョブの識別子
      name: "Daily Analysis Job"
      tasks:
        - task_key: "run_notebook"
          notebook_task:
            notebook_path: "./notebook.ipynb"

      # ここが permissions の階層
      permissions:
        - group_name: "data-engineers"
          level: "CAN_MANAGE"
        - group_name: "analysts"
          level: "CAN_VIEW"

Saltingはどういう操作で、どういう時に有効？

Databricks（Apache Spark）におけるSalting（ソルティング）は、データの偏り（データスキュー / Data Skew）を解消し、並列処理のボトルネックを打破するための高度なチューニング手法です。

Salting の概要

1. どんな時に有効か？（スキューの検知）

以下のような症状が出ている時に非常に有効です。

• 99%のタスクは数秒で終わるのに、最後の1つのタスクだけが数分〜数時間終わらない。
• Spark UI の「Executor」タブで、特定のコアだけがずっと稼働しており、他が遊んでいる。
• 特定のキー（例：user_id = 0 や country = 'US'）のレコード数が数千万件ある。

Saltingはどうすれば行える？

以下のような形

from pyspark.sql import functions as F

# 塩の範囲（パーティションをいくつにバラしたいか）
salt_bins = 5

# テーブルA: 結合キーに 1〜5 のランダムな数値を付加
df_skewed = df_a.withColumn(
    "salted_key", 
    F.concat(F.col("user_id"), F.lit("_"), F.expr(f"int(rand() * {salt_bins})"))
)

様々なフィルター条件を使われる場合、DeltaよりHiveの方が良いケースがある？

結論から言うと、現代のDatabricks環境において「Hiveの方が良いケース」はほぼ存在しません。

特に「様々なフィルター条件（WHERE句）」が使われるアドホックなクエリや複雑なワークロードにおいては、Delta Lakeの方が圧倒的に有利です。

試験対策として、なぜHive（標準的なParquetテーブル）ではなくDeltaが選ばれるのか、フィルター性能の観点から比較表にまとめました。

Delta Lake vs Hive (Parquet) のフィルター性能比較

2. 多次元的な最適化 (Z-Order / Liquid Clustering)

Hiveのパーティションは通常1つか2つのカラムに限られます。

• Deltaの場合: Liquid Clustering を使えば、多数のカラムに対して柔軟にクラスタリングをかけられます。ユーザーが「日付で絞る」「顧客IDで絞る」「商品カテゴリで絞る」といったバラバラな条件で検索しても、どのパターンでも高速に応答できます。

3. ブルームフィルタ・インデックス (Bloom Filter Index)

特定の高カーディナリティなカラム（例：IDやシリアル番号）で完全一致検索（=）を多用する場合、Deltaではブルームフィルタを作成できます。これにより「そのファイルに探している値が含まれていないこと」を瞬時に判断できます。

Liquid Clusteringって記憶ではパーテーションキーとソートキーは固定しないと使えないはずだが、それでどうやって柔軟な検索が可能になる？

実は、Liquid Clusteringの最大の武器は「パーティションキーという概念を捨て、カラム（キー）を後から自由に入れ替えられる柔軟性」にあります。

なぜ「固定」ではなく「柔軟」なのか？

従来の方式（HiveパーティショニングやZ-Order）と比較して、Liquid Clusteringがなぜ柔軟と言われるのか、その理由を整理します。

Liquid Clusteringでは、以下のコマンドで運用中にキーを変更できます。

- テーブル作成後、やっぱり別のカラムで最適化したくなった場合
ALTER TABLE table_name CLUSTER BY (new_column_1, new_column_2);

このコマンドを打った後の OPTIMIZE 実行時から、新しいキーに基づいてデータが再配置され始めます。「テーブル定義を壊さずに、ビジネス要件に合わせて最適化の軸を変えられる」のが最大の柔軟性です。

2. 「多次元」での高速検索

Z-Orderも多次元検索に強いですが、データが増えるほど「並べ替えコスト」が指数関数的に増大します。

Liquid Clusteringは、「ヒルベルト曲線」などの高度なアルゴリズムを内部で利用し、複数のカラム（最大4つ推奨）の相関関係を保ったままデータを物理的にまとめます。これにより、「日付のみ」「IDのみ」「日付＋ID」といった、どのパターンのフィルター条件が来ても、関連するデータブロックだけを的確に狙い撃ち（Data Skipping）できます。

3. パーティションの「細分化」からの解放

「日付」でパーティションを切ると、1日のデータが少ない場合にファイルが細かくなりすぎて性能が落ちました。
Liquid Clusteringは「物理的なディレクトリ構造」に依存しないため、データ量に応じて自動でクラスターのサイズを調整します。これにより、設計者が「どの粒度でパーティションを切るべきか」と悩む必要がなくなりました。

試験対策のポイント

試験で「Liquid Clusteringの利点」を問われたら、以下のキーワードが正解に直結します。

• No Partition Evolution required: パーティションの再定義（マイグレーション）が不要。
• Support high-cardinality columns: 顧客IDのようなユニークな値が多いカラムでも性能劣化しない。
• Incremental Clustering: OPTIMIZE を実行するたびに、新しく入ってきたデータだけを賢くクラスタリングしてくれる。

Delta Live TablesとStructured Streamingのメリデメ表

例えるならば、Structured Streamingは「エンジン（部品）」であり、DLTはそのエンジンを積んだ「自動運転車（フレームワーク）」です。

DLT vs Structured Streaming 比較表

1. Structured Streaming のメリット・デメリット

• メリット:
  • 究極の柔軟性: トリガーの間隔、チェックポイントの場所、状態管理（Stateful processing）を細かく制御できる。
  • 既存コードの流用: 純粋なSparkコードなので、既存のライブラリや処理ロジックをそのまま組み込みやすい。
• デメリット:
  • 運用負荷: チェックポイントの管理や、クラスターが落ちた際の再起動、スキーマ進化の対応などを自分で行う必要がある。
  • 可視性の欠如: データの流れ（リネージ）を追うのが難しく、パイプライン全体の状態把握に手間がかかる。

2. Delta Live Tables (DLT) のメリット・デメリット

• メリット:
  • 宣言的開発: 「どう動かすか」ではなく「どんなテーブルを作りたいか」を書くだけで、システムが最適化してくれる。
  • データ品質 (Expectations): expect_or_drop などを使って、不正データを自動で除外・監視できる。
  • 自動化: クラスターのサイズ調整（Enhanced Autoscaling）や、エラー時の自動リトライが標準搭載。
• デメリット:
  • 制約: DLTのフレームワーク内で許可されていない操作（特定のライブラリの使用や複雑なI/O）が制限されることがある。
  • 学習コスト: DLT独自の構文（live. プレフィックスなど）やライフサイクルを理解する必要がある。

Delta Live TablesとStructured Streamingを併用したコード例

import dlt
from pyspark.sql import functions as F

# --- 1. Bronze層: 外部ソースから Structured Streaming で取り込み ---
# spark.readStream + format("cloudFiles") を使用
@dlt.table(
    name="raw_game_logs",
    comment="Auto Loaderを使用してクラウドストレージから生ログを取り込む"
)
def raw_game_logs():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .load("/mnt/data/events/")
    )

# --- 2. Silver層: DLT間のストリーミング接続とデータ品質管理 ---
# dlt.read_stream を使用して、上流のテーブルから増分を読み出す
@dlt.table(
    name="cleaned_game_logs",
    comment="不正なスコアを除去し、タイムスタンプを正規化する"
)
@dlt.expect_or_drop("valid_score", "score >= 0") # データ品質の制約
def cleaned_game_logs():
    return (
        dlt.read_stream("raw_game_logs") # 前のテーブルをストリームとして参照
        .withColumn("event_timestamp", F.to_timestamp("event_time"))
        .select("user_id", "event_type", "score", "event_timestamp")
    )

# --- 3. Gold層: マテリアライズドビューによる集計 ---
# 集計（Window関数など）を行うため、ここではストリームではなくテーブルとして定義
@dlt.table(
    name="daily_score_summary",
    comment="ユーザーごとの日次合計スコア"
)
def daily_score_summary():
    return (
        dlt.read("cleaned_game_logs") # 集計の場合は read_stream ではなく read を使うのが一般的
        .groupBy("user_id", F.window("event_timestamp", "1 day"))
        .agg(F.sum("score").alias("total_daily_score"))
    )

Bronze層で expect（警告のみ）を設定しつつ、その違反レコードをSilver層で「後追い」で除外したいというシナリオ

最もスマートな方法は

「Bronzeは全件入れておき、Silverの定義で expect_or_drop を使う」

という構成です。これにより、Bronzeには調査用の「ダメなデータ」を残しつつ、分析用のSilverはクリーンに保てます。

spark.readStream と dlt.read_stream の違い

1. spark.readStream：外部からの「取り込み」

spark.readStream は、標準的な Apache Spark の命令です。DLT の外にある「ファイル」や「メッセージキュー」からデータを吸い上げる時に使います。

• いつ使う？: パイプラインの一番最初（通常は Bronze 層）で、クラウドストレージ上の RAW データを読み込むとき。
• 特徴: cloudFiles (Auto Loader) などのフォーマットを指定して、外部との接続を確立します。

2. dlt.read_stream：内部での「参照」

dlt.read_stream は、DLT フレームワーク専用の特別な関数です。

• いつ使う？: すでに DLT 内で定義した別のテーブル（例：bronze_table）から、次のテーブル（例：silver_table）へデータを流したいとき。
• 最大のメリット（自動DAG構築）:
  この関数を使うことで、Databricks は「テーブル A はテーブル B の前に実行しなければならない」という依存関係（DAG: Directed Acyclic Graph）を自動で理解します。

cloudFiles (Auto Loader) ってなに

cloudFiles=Auto Loader?

結論から言うと、「概念（機能名）としては Auto Loader」であり、「コード上の指定（識別子）としては cloudFiles」です。

TRANSFORMとexpect actualでテストする場合の具体例

具体的なコード例

from pyspark.sql import DataFrame
import pyspark.sql.functions as F

# 1. 変換ロジックを独立した関数として定義（再利用・テスト可能）
def add_tax_transform(df: DataFrame) -> DataFrame:
    return df.withColumn("total_price", F.col("price") * 1.1)

def filter_invalid_orders(df: DataFrame) -> DataFrame:
    return df.filter(F.col("price") > 0)

# 2. 実際のジョブやDLTでの使用例（メソッドチェーン）
# transform() を使うことで、上から下へ流れるように書ける
df_result = (
    spark.read.table("bronze_orders")
    .transform(filter_invalid_orders)
    .transform(add_tax_transform)
)

# --- テストコードのイメージ ---

# 1. テスト用データの作成 (Actual)
input_df = spark.createDataFrame([(100,), (200,)], ["price"])

# 2. ロジックの適用
actual_df = input_df.transform(add_tax_transform)

# 3. 期待するデータの作成 (Expected)
expected_df = spark.createDataFrame([(100, 110.0), (200, 220.0)], ["price", "total_price"])

# 4. 比較（一致すればテスト合格）
assert actual_df.collect() == expected_df.collect()

多様な時間でフィルターする場合は、時間はISO 8601形式の文字列で保存で充分？それとも値？

「文字列（ISO 8601）」ではなく、必ず「Timestamp型（内部的にはLong値）」で保存すべきです。

なぜ「数値（Timestamp型）」の方がフィルターに強いのか？

1. Delta Lake の「Data Skipping」

Delta Lake は、ファイルの各カラムの Min（最小値） と Max（最大値） を統計情報として保持しています。

• Timestamp型の場合: 数値として管理されているため、WHERE time > '2026-03-01' というクエリに対し、エンジンが「このファイルの最大値は 2026-02-28 だから読む必要なし」と瞬時に判断（スキップ）できます。
• 文字列の場合: 文字列としての Min/Max 比較になるため、型に比べるとオーバーヘッドが大きく、複雑な時間計算（「1時間前」など）を含むフィルターの最適化が困難になります。

2. パーティショニングと Z-Order

時間をフィルターの主軸にする場合、Date 型に変換してパーティショニングしたり、Timestamp 型で Z-Order や Liquid Clustering を適用したりします。
これらは数値的な順序に基づいてデータを物理的に並べ替えるため、型が一致していないと十分な性能を発揮できません。

Timestamp型を効率的に、かつ多様な条件でフィルターする際のSQL例

-- 特定の期間を指定
SELECT * FROM orders 
WHERE order_timestamp BETWEEN '2026-01-01 00:00:00' AND '2026-01-31 23:59:59';

-- 単純な比較演算子
SELECT * FROM orders 
WHERE order_timestamp >= '2026-03-01';

Shuffle Partitionとは何？

なぜ「シャッフル」が必要なのか？

例えば、各ノードにバラバラに散らばっている「売上データ」を「地域ごと」に集計したい場合、同じ地域のデータを一つの場所に集める必要があります。この 「データをネットワーク越しに移動させて再配置するプロセス」がシャッフルです。

このとき、移動したデータをいくつの塊（パーティション）に分けるかを決めるのが Shuffle Partition です。

Liquid Clusteringってパーテーションキーとソートキーは途中でいくらでも変えれる？

はい、いつでも、何度でも変更可能です。

Delta Sharingに関するopen share機能って静的Deltaテーブルのみ共有可能？ノートブックなどは？

Open Sharingはdatabricks外への共有を行うもの

Open Sharingの手順

1. アクティベーションリンクの送付:
   データ提供者が共有設定をすると、一回限りの「アクティベーションリンク（URL）」が発行されます。これを相手にメール等で送ります。
2. 資格情報ファイル（JSON）のダウンロード:
   相手がそのリンクをクリックすると、資格情報ファイル（config.share）というJSONファイルがダウンロードできます。
   • 注意: このリンクには有効期限があり、一度しかダウンロードできません。
3. オープンソースクライアントでの接続:
   相手は、Python、Pandas、Power BI、Apache Spark、TableauなどのDelta Sharing対応クライアントにそのJSONファイルを読み込ませることで、あたかも自分の手元にデータがあるかのようにクエリを実行できます。

データを保存する場合、UCのマスク関数では、保存されるデータはマスクされない問題がある？

はい、その通りです。

Unity Catalog（UC）の「カラムマスク（Masking Functions）」は、「読み取り時（クエリ実行時）」に動的にデータを隠す機能であり、ストレージ（S3やADLS上のDeltaファイル）に保存されている物理データそのものを書き換えるわけではありません。

マテリアライズドビューとはどんなもの？

Materialized Viewは、簡単に言うと「クエリの結果をあらかじめ計算して、物理的に保存しておくテーブル」のことです。

コード比較例

通常のビュー

-- 定義するだけ（データは保存されない）
CREATE VIEW main.default.high_value_orders_view
AS
SELECT 
  order_id, 
  customer_id, 
  amount 
FROM main.default.orders
WHERE amount > 1000;

マテリアライズドビュー

-- データを計算して物理保存する
CREATE MATERIALIZED VIEW main.default.high_value_orders_mv
AS
SELECT 
  order_id, 
  customer_id, 
  amount 
FROM main.default.orders
WHERE amount > 1000;

-- 最新状態にするには「リフレッシュ」が必要
REFRESH MATERIALIZED VIEW main.default.high_value_orders_mv;

DLT版 通常のビュー

import dlt
@dlt.view
def temporary_filter_view():
    # パイプライン内でのみ有効な一時的なビュー
    # カタログには保存されない
    return dlt.read("raw_data").filter("id IS NOT NULL")

DLT版 マテリアライズドビュー

import dlt

@dlt.table(
  name="sales_summary_mv",
  comment="集計結果を保持するマテリアライズドビュー"
)
def sales_summary():
    # dlt.read() を使うことで、上流の全データを読み込んで集計し、
    # その結果を物理テーブルとして保存する
    return (
        dlt.read("cleaned_sales") 
        .groupBy("region")
        .sum("amount")
    )

具体的な挙動のイメージ

例えば、数億件の「注文ログ」から「日別の売上集計」を出す場合：

1. 作成: CREATE MATERIALIZED VIEW daily_sales AS SELECT date, sum(amount) FROM orders GROUP BY date;
2. 物理保存: Databricks はこの集計結果を Delta テーブルとして保存します。
3. 参照: ユーザーが SELECT * FROM daily_sales と打つと、数億件の計算を飛ばして、集計後の数千行を読み込むだけになります。
4. リフレッシュ: 新しい注文データが入ってきたら、REFRESH MATERIALIZED VIEW daily_sales を実行（または自動スケジュール）して中身を最新にします。

試験・実務でのポイント

• 「いつ MV を使うべきか？」:
  • ベーステーブルが巨大。
  • 変換処理（Transform）が重い。
  • クエリの頻度が高い（BIダッシュボードなど）。
• 「最新性」の妥当性:
  • MV はリフレッシュされるまでデータが古くなる可能性があります。秒単位のリアルタイム性が必要な場合は、通常のビューやストリーミングテーブルを検討します。
• Unity Catalog での管理:
  • 最近の Databricks では、Unity Catalog 上で作成する MV が推奨されています。これにより、他の BI ツールからも高速な MV にアクセスできるようになります。

マテリアライズドビューは複雑な財務処理関数などと相性いい？

「非常に相性が良い」です。

複雑な財務処理（外貨換算、複雑な償却計算、税額算出など）は計算コストが高く、かつコードの保守性が求められるため、マテリアライズドビュー（MV）を活用するメリットが最大化されます。

計算コストの「事前払い」ができるため

列のマスクを、ユーザーグループごとによってUDFを使って実現する場合、どんな実装になる？

1. マスク用関数（SQL UDF）の作成

CREATE OR REPLACE FUNCTION email_mask(email STRING)
RETURN CASE
  -- HR（人事）グループならそのまま見せる
  WHEN is_account_group_member('hr_users') THEN email
  -- 一般ユーザーならドメイン以外を隠す（例: a***@example.com）
  WHEN is_account_group_member('standard_users') THEN regexp_replace(email, '^.*(?=@)', '***')
  -- それ以外（外部の人など）は完全に隠す
  ELSE 'REDACTED'
END;

1. テーブルのカラムにマスクを適用する

-- テーブル作成時に指定する場合
CREATE TABLE customers (
  user_id INT,
  email STRING MASK email_mask,  -- ここで関数をバインド
  full_name STRING
);

-- 既存のテーブルに後から適用する場合
ALTER TABLE customers ALTER COLUMN email SET MASK email_mask;

CatalogレベルでUCで権限設定してる場合、現状あるテーブルや、以降作る配下のスキーマなどの権限はどうなる？カスケード（継承される）？

はい、Unity Catalog（UC）の権限は「カスケード（継承）」されます。

1. メタストア (Metastore)
   └── 2. カタログ (Catalog)  ← ここで権限を振ると...
       └── 3. スキーマ (Schema)  ← 自動で継承
           └── 4. テーブル / ビュー / ボリューム  ← 自動で継承

3. 注意点：継承の「上書き」や「拒否」はできる？

実務や試験で混乱しやすいポイントがいくつかあります。

• DENY は継承よりも優先されます。例えば、カタログレベルで SELECT 権限があっても、特定のスキーマに対して DENY SELECT を設定すれば、そのユーザーはそのスキーマを見ることができなくなります。
• 「例外的に一部のスキーマだけ隠す」は難しい:
  カタログレベルで SELECT を与えてしまうと、そのカタログ内の特定のスキーマだけを見せないようにすることはできません。
  • 対策: 権限を絞りたい場合は、カタログレベルではなく、個別のスキーマレベルで権限を付与するのがベストプラクティスです。
• Ownership（所有権）の継承:
  所有権（OWNER）は継承されません。カタログの所有者であっても、他人が作ったスキーマの所有者にはなりませんが、カタログレベルの強力な権限（CREATE など）によって管理は可能です。

Auto loaderで、imageや、binary、textって括りはある？

Auto Loader（cloudFiles）において、「image」や「binary」といった専用のフォーマット指定（括り）はありません。

Auto Loaderが直接サポートしている「ファイル形式」は、あくまで構造化・半構造化データ（JSON, CSV, Parquet等）が中心です。しかし、画像やバイナリファイルを扱いたい場合には、binaryFile という形式を組み合わせて使うのが一般的です。

SCD Type 1 vs. Type 2ってなに

データエンジニアリングの世界で避けては通れないのが、この SCD（Slowly Changing Dimension：ゆっくり変化する次元） です。

一言で言うと、「マスタデータの値が変わったとき、過去の履歴をどう扱うか？」 という戦略の違いです。

SCD Type 1：上書き（履歴を残さない）

「過去なんて振り返らない」スタイルです。既存のレコードを新しい値でそのまま上書きします。

• 挙動: UPDATE 処理のみ。
• 用途: 名前のタイポ修正や、過去の値を保持する必要がまったくない項目（例：Emailアドレスの変更など）。
• メリット: データ量が最小限で済み、クエリが非常にシンプル。
• デメリット: 「1ヶ月前はどうだったか？」という分析が不可能になる。

SCD Type 2：履歴保持（新しい行を追加）

「過去も大切にする」スタイルです。値が変わるたびに新しい行を追加し、どのデータが「いつからいつまで有効だったか」を管理します。

• 挙動: 既存の行を「古い」とし、新しい行を INSERT する。
• 管理カラム: 通常、start_date（有効開始日）、end_date（有効終了日）、is_current（最新フラグ）を追加します。
• 用途: 住所（地域別の売上分析に必要）、役職、製品価格など、時間の経過に伴う変化を追いたい項目。
• メリット: 過去の任意の時点の状態を正確に再現できる。
• デメリット: データ量が肥大化し、結合（Join）やクエリが複雑になる。

CDC データってなに

DC (Change Data Capture) とは、日本語で「変更データキャプチャ」と呼ばれ、データベースに加えられた「追加（Insert）」「更新（Update）」「削除（Delete）」の変更履歴だけを抽出して、別のシステムに同期させる手法のことです。

これまでに話した「Auto Loader」や「SCD Type 2」と非常に関連が深い技術です。

なぜ CDC が必要なのか？

例えば、基幹システムの MySQL にある「顧客テーブル」を Databricks に同期したいとします。

• 従来の方法（フルロード）: 毎日、数千万件の全データを引っこ抜く。
  • → データベースに負荷がかかりすぎるし、時間もかかる。
• CDC の方法: 「今日の10時に Aさんの住所が変わった」というログだけを追いかける。
  • → 負荷が極めて低く、ほぼリアルタイムで同期できる。

CDC データの構造（イメージ）

CDC ツール（Debezium など）を通すと、データは通常以下のような「メタデータ付きの形式」で届きます。

Databricks における CDC の強力な味方

CDC データは「単なる変更の羅列」なので、そのままではテーブルとして使えません（田中さんの最新住所は？という問いに答えにくい）。そこで、以下の機能を使います。

1. APPLY CHANGES INTO (Delta Live Tables)

これが最強のツールです。CDC で届いた「I, U, D」のフラグを読み取り、ターゲットの Delta テーブルに対して SCD Type 1 や Type 2 を自動適用してくれます。

2. MERGE 文

SQL や Python で手動で CDC データを反映させる場合は、MERGE 文を使います。「マッチしたら Update、なければ Insert」という処理を 1 つのトランザクションで行います。

Databricks（Delta Lake）でSCDやCDCを扱う際の、「現場でそのまま使えるSQL」を整理します。

主に、手動で実行する MERGE INTO と、Delta Live Tables (DLT) で自動化する APPLY CHANGES INTO の2つのパターンがあります。

1. 手動で行う：MERGE INTO (SCD Type 1)

「既存のデータがあれば更新、なければ追加」という最も一般的な操作です。

MERGE INTO customers AS target
USING customer_updates AS source
ON target.user_id = source.user_id
-- 1. マッチした場合：新しい値で上書き（SCD Type 1）
WHEN MATCHED THEN
  UPDATE SET 
    target.email = source.email,
    target.address = source.address,
    target.updated_at = source.updated_at
-- 2. マッチしなかった場合：新規作成
WHEN NOT MATCHED THEN
  INSERT (user_id, name, email, address, updated_at)
  VALUES (source.user_id, source.name, source.email, source.address, source.updated_at);

2. DLTで自動化する：APPLY CHANGES INTO

CDCデータを処理する場合、この構文を使うのがDatabricksのベストプラクティスです。

SCD Type 1 の場合（上書き）

CREATE OR REFRESH STREAMING TABLE target_table;
APPLY CHANGES INTO LIVE.target_table
FROM STREAM(LIVE.source_cdc_stream)
KEYS (user_id) - 主キー
SEQUENCE BY updated_at - 順序保証用のカラム
COLUMNS  EXCEPT (event_type) - 除外するカラムがあれば指定
STORED AS SCD TYPE 1; - Type 1（履歴なし）を指定

SCD Type 2 の場合（履歴保持）

これだけで、__start_at や __end_at といった履歴管理用のカラムが自動生成されます。

APPLY CHANGES INTO LIVE.target_table
FROM STREAM(LIVE.source_cdc_stream)
KEYS (user_id)
SEQUENCE BY updated_at
STORED AS SCD TYPE 2;                  -- Type 2（履歴保持）を指定

Change Data Feedを実際に使う例

Change Data Feed (CDF) は、Deltaテーブルに加えられた「変更そのもの（インサート、更新前の値、更新後の値、削除）」を、ログとして抽出できる機能です。

単に「最新の状態」を見るのではなく、「何がどう変わったか」という履歴を、下流の処理に効率的に伝えるために使われます。

CDF を使うための 3 ステップ

1. テーブルで CDF を有効にする

- テーブル作成時に有効化
CREATE TABLE silver_users ( id INT, name STRING, city STRING
)
TBLPROPERTIES (delta.enableChangeDataFeed = true);

- 既存のテーブルで有効化
ALTER TABLE silver_users SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

2. データの変更操作を行う

通常通り、INSERT や UPDATE を行います。

SQL

UPDATE silver_users SET city = 'Tokyo' WHERE id = 101;

3. 変更点だけを読み出す (SQL / Python)

ここが CDF の真骨頂です。特定のバージョン間や時間帯の「差分」を指定して取得できます。

SQL の場合:

- バージョン1から最新までの「変更履歴」を取得

SELECT  FROM table_changes('silver_users', 1);
- 特定の時間範囲で取得

SELECT  FROM table_changes('silver_users', '2026-03-04 10:00:00', '2026-03-04 12:00:00');

ウォーターマークとは、使用例

ウォーターマーク（Watermark）とは、ストリーミング処理において「遅れて届いたデータをどこまで待つか」を定義する境界線のことです。

from pyspark.sql import functions as F

df = (spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "json")
.load("/datapath"))

ウォーターマークの設定：event_timeを基準に、5分間の遅延を許容

windowed_counts = (df
.withWatermark("event_time", "5 minutes")
.groupBy(
F.window("event_time", "10 minutes")
)
.count())

この設定での挙動

1. 20:00 〜 20:10 のウィンドウが集計中。
2. 20:05 に発生したデータが、20:09 に届いた → 集計に含まれる。
3. 20:02 に発生したデータが、20:12 に届いた → 5分以上の遅れだが、ウォーターマークがまだ更新されていなければ入る可能性がある（※詳細後述）。
4. システムが 20:15 のデータを処理し、ウォーターマークが 20:10 まで進んだ後、20:04 のデータが届いた → 完全に無視される。

ウォーターマーク、with句、windowなど、必要なものを用いて、以下のアラートを生むSQL

条件

・1日一回アラート

・以降はアラート条件の時は毎日アラート

・毎日何回もアラートは出ない

・条件は3日連続平均気温が5度以上の時

-- 1. 日次の平均気温を算出するストリーミングテーブル（Bronze -> Silver）
-- ここで1日1行のデータに集約し、ウォーターマークで古いデータを管理します
WITH daily_summary AS (
  SELECT
    window.start AS alert_date,
    AVG(temperature) AS avg_daily_temp
  FROM STREAM(sensor_data)
  -- データの遅延を最大1日許容
  WATERMARK event_time DELAY '1 day'
  -- 1日単位でウィンドウ集計（これで1日1件のレコードになる）
  GROUP BY window(event_time, '1 day')
),

-- 2. 3日間の移動平均を計算する（Silver -> Gold相当）
moving_average_stats AS (
  SELECT
    alert_date,
    avg_daily_temp,
    -- 過去2行＋現在行（計3日）の平均を算出
    AVG(avg_daily_temp) OVER (
      ORDER BY alert_date 
      ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
    ) AS three_day_avg,
    -- 3日分のデータが揃っているかチェックするためのカウント
    COUNT(avg_daily_temp) OVER (
      ORDER BY alert_date 
      ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
    ) AS day_count
  FROM daily_summary
)

-- 3. 最終的なアラート判定
SELECT
  alert_date,
  three_day_avg,
  'ALERT: 3-day consecutive average temperature >= 5.0' AS alert_message
FROM moving_average_stats
WHERE 
  -- 3日間の平均が5度以上
  three_day_avg >= 5 
  -- 運用開始直後（1日目や2日目）の誤検知を防ぐため、3日分揃っていることを確認
  AND day_count = 3

Coalesce vs Repartition:の違い。まずCoalesce(1)とは何

Coalesce(1) は、一言で言えば「バラバラに散らばったデータを、無理やり1つの箱（パーティション）に凝縮する」操作です。

データエンジニアリングの実務において、Coalesce(1) が登場するシーンはほぼ一つ：「最終結果を1つのファイルとして出力したい時」です。

1. Coalesce(1) とは何をしているのか？

Apache Sparkにおいて、データは通常複数のワーカーノードに分散して保持されています。Coalesce(1) を実行すると、Sparkはネットワーク経由で全データを1つのノードに集め、1つのパーティションにまとめます。

最大の特徴：シャッフルを「最小限」に抑える

Repartition との決定的な違いは、フルシャッフル（全データの無差別な再配置）を避ける点です。Coalesce は既存のパーティションを「結合」するだけなので、データの移動を最小限に抑えようとします。

3. なぜ Coalesce(1) を使うのか？（実務例）

一番多いのは、外部システム（Excelで読み込むCSVや、特定のツールが読み込む設定ファイルなど）が「1つのファイルしか受け付けない」場合です。

# 何百個もファイルができるのを防ぎ、result.csv 1つだけを作る
df.coalesce(1).write.format("csv").save("/path/output")

Wheel Fileを、インターネット経由なしで活用したい場合、どんな感じでインストールするのが望ましい？

インターネットに接続できない環境（エアギャップ環境や厳しいプロキシ制限下）でPythonの Wheelファイル (.whl) を活用する場合、Databricksでは 「Unity CatalogのVolumes」 または 「Workspace Files」 にファイルを配置してインストールするのが現代的で最もスマートな方法です。

環境や用途に合わせて、望ましい3つのアプローチを解説します。

1. Unity Catalog Volumes を使う (推奨：最もセキュアで近代的)

Unity Catalogが有効な環境であれば、Volumes（ボリューム）を「社内限定のアーティファクト置き場」として使うのがベストです。

• 手順:
  1. UIまたはCLIで、ローカルの .whl ファイルを Volumes のパス（例: /Volumes/main/default/my_lib_vol/）にアップロードします。
  2. ノートブックの先頭で以下を実行します。Python

     %pip install /Volumes/main/default/my_lib_vol/my_package-0.1-py3-none-any.whl

• メリット: 権限管理（GRANT）が効くため、特定のチームだけにライブラリの使用を許可できます。

複雑な財務処理UDF × Pandas on Spark の上手い使い方

各グループ（例：国別、通貨別）ごとに複雑な計算を行う場合、applyInPandas (Grouped Map UDF) を使うのがベストです。

• 理由: 通常のUDFは1行ずつ処理するため遅いですが、applyInPandas はグループ単位でデータを Pandas DataFrame としてまとめて Python プロセスに渡し、ベクトル演算（一括計算）を行います。
• メリット: 財務ライブラリ（numpyや既存のPythonロジック）をそのまま使いつつ、Sparkで並列処理ができます。

クエリの実行時間はどこで見るべき？

• Query Profile (SQLブラウザ): 「どの演算（JoinやFilter）に時間がかかったか」「インデックス（Data Skipping）が効いているか」を視覚的に見る場所です。SQLチューニングならここ一択です。
• ジョブの履歴画面 / Spark UI: 「特定のタスクが偏っていないか（スキュー）」「ネットワーク転送（シャッフル）が重くないか」など、インフラ側の挙動を確認するのに適しています。

Shuffle Hash Join とは

結合する両方のテーブルを、結合キーに基づいて全ノードにシャッフル（再配置）し、各ノードでハッシュテーブルを作成して結合する手法です。

• 使い時: Shuffle Hash Joinは、「Sort Merge Joinを実行するには各パーティション内のソート負荷が高すぎる場合」、かつ「片方のテーブルの各パーティションが、Executorのメモリにハッシュテーブルとして収まるサイズである場合」に有効です。
  一般的に、非常に巨大なテーブル同士の結合は Sort Merge Join が最も安定します。

データのスキュー回避策

• maxPartitionBytes の調整: 1パーティションあたりのサイズを小さくして、タスクをより細かく分散させるのは有効です。
• インスタンス強化: メモリを増やすと OOM（メモリ不足）は防げますが、スキュー（1つのコアだけ頑張っている状態）の根本解決にはなりません。
• Salting (ソルティング): 結局、これが最も根本的な解決策です。

タグをつけるコマンド

テーブルやカラムにタグを付けるのは ALTER TABLE コマンド（SQL）で行います。

ALTER TABLE table_name SET TAGS ('department' = 'finance', 'priority' = 'high');

※ sql.conf はセッションごとの設定変更に使うもので、メタデータ（タグ）の付与には使いません。

AIがカラムの説明を自動生成する機能はある？

あります。 Unity Catalog（UC）の機能として、「AI-generated documentation」が提供されています。
カタログエクスプローラー上で「Generate」ボタンを押すと、テーブル名やデータの中身をAIが推論し、カラムの説明（Description）を自動で下書きしてくれます。

DatabricksのAI-generated documentationについては、テーブルにカラムが追加される？

テーブルに新しいカラム自体が物理的に追加されることはありません。

AIが生成したテキストは、Unity Catalog上の以下のフィールドに書き込まれます。

• Table Comment: テーブル全体の概要。
• Column Comment: 各カラムの意味や内容の説明。

Delta は外部テーブルの最新情報を保証する？

• Managed Table（管理テーブル）: Databricksが完全に管理するため、常に最新です。
• External Table（外部テーブル）: Delta形式であれば、他のエンジンで書き込まれても VACUUM や OPTIMIZE の整合性は保たれますが、非Delta（Parquet等）の外部ファイルを直接参照している場合は、REFRESH TABLE を実行するまで新しく追加されたファイルが見えないことがあります。

Standard と Dedicated みたいな設定はある？

ない。正しくは以下

• Serverless: リソース管理が不要で、すぐに起動する（推奨）。
• Pro: ワークフローなどに適した標準的な構成。
• Classic: 最も基本的な構成（古い機能）。

まず@dltってコマンドは何を意味する？

Databricksのデータエンジニアリングにおいて、@dlt（またはPythonのデコレータとしての @dlt.table / @dlt.view）は、Delta Live Tables (DLT) というパイプライン専用のフレームワークで使われる宣言的な修飾子を意味します。

簡単に言うと、「これから書く関数（またはSQL文）は、ただのクエリではなく、DLTパイプラインの一部として管理されるテーブルだよ」とシステムに教える合図です。

@dlt が持つ主な役割

DLTパイプラインを実行する際、このデコレータがあることで以下の処理が自動化されます。

1. 依存関係の自動解決 (DAGの構築):
   どのテーブルがどのテーブルを参照しているかを解析し、正しい順番でデータを処理する「実行計画（グラフ）」を自動で作ってくれます。
2. スキーマの自動管理:
   データの構造が変わっても、ターゲットとなるDelta Tableのスキーマを自動的に適用・更新します。
3. チェックポイントとリトライ:
   処理が途中で止まっても、どこまで進んだかを記録しているため、最初からやり直す必要がありません。

import dlt

@dlt.table(
  name="raw_player_logs",
  comment="ゲームの生ログデータ"
)
def raw_player_logs():
    return spark.readStream.format("cloudFiles").load("/path/to/logs")

CREATE OR REFRESH LIVE TABLE raw_player_logs
AS SELECT * FROM read_files("/path/to/logs")

CREATE OR REFRESH LIVE TABLE raw_player_logsってなに

SQLにおける CREATE OR REFRESH LIVE TABLE は、Delta Live Tables (DLT) というフレームワーク特有の構文です。

通常のSQL（Spark SQL）の CREATE TABLE とは決定的に違う、「宣言型（Declarative）」という考え方が反映されています。一言でいうと、「このテーブルを常に最新の状態に保て（手順はお任せする）」という命令です。

それぞれの単語の意味を解剖してみましょう。

1. 各キーワードの役割

• CREATE OR REFRESH
  「テーブルがなければ作る。すでにあるなら、中身を最新のデータに更新（リフレッシュ）する」という意味です。手動で INSERT や UPDATE を書く必要がなくなります。
• LIVE
  これが最も重要です。このテーブルが DLTパイプラインの一部 であることを示します。通常のテーブルとは違い、背後で「依存関係の管理」や「自動リトライ」が行われる特別な存在になります。
• TABLE
  結果を物理的な Delta Table として保存します（対照的に、一時的な計算結果なら LIVE VIEW を使います）。

2. なぜこれを使うのか？（データエンジニアのメリット）

データエンジニアとしての実務（Databricks利用時）では、以下のような面倒な作業がすべて自動化されます。

1. 増分更新の自動化: ソースデータに新しいログ（ゲームのプレイ記録など）が追加されたとき、前回の続きから自動で読み込んでくれます。
2. 依存関係の解決: 例えば「Goldテーブル」が「Silverテーブル」を参照している場合、Silverが更新された後にGoldを動かす、といった実行順序をDLTが勝手に判断します。
3. スキーマの進化: ゲームのアップデートでログのカラムが増えた場合も、オプション設定（Schema Evolution）次第で自動対応できます。

spark.readStream.format(“cloudFiles”)についてcloudFiles以外のオプションは？

WINDOW関数と.windowは別物？

ややこしいがF.window()というまた別のものもある。

1. SQL の WINDOW 関数

標準的なSQLで、特定の範囲（パーティション）に対して計算を行うための構文です。

• 記述場所: SELECT 文の中。
• 特徴: OVER (PARTITION BY ... ORDER BY ...) という句を使います。

SELECT 
  user_id, 
  score,
  -- これがSQLのWINDOW関数
  AVG(score) OVER (PARTITION BY user_id ORDER BY date) as rolling_avg
FROM game_logs

2. DataFrame API の .window() （PySpark）

Pythonのコード内で、SQLの OVER 句と同じ「計算の枠組み」を定義するためのメソッドです。

• 記述場所: Python コード（pyspark.sql.Window モジュール）。
• 特徴: まず「窓の定義」を作り、それを over() メソッドに渡します。

from pyspark.sql import Window
import pyspark.sql.functions as F

# 1. 窓（WindowSpec）を定義する
w = Window.partitionBy("user_id").orderBy("date")

# 2. 関数に適用する
df.withColumn("rolling_avg", F.avg("score").over(w))

はい、時間あたりのグルーピングに関しては、F.window() の方が圧倒的に便利で強力です。

1. F.window() の具体的なイメージ

import pyspark.sql.functions as F

# 10分間隔でデータを集約する
df_aggregated = df.groupBy(
    F.window(F.col("event_time"), "10 minutes")
).agg(
    F.count("user_id").alias("login_count")
)

シナリオ：ゲームログのクレンジング

以下の3つの処理を行いたいとします。

1. テストユーザーの除外（filter_test_users）
2. スコアの正規化（normalize_score）
3. 不正な移動速度のフラグ立て（add_cheater_flag）

sparkのtransformがあると便利な例

1. .transform() を使わない場合（変数地獄）

中間変数がたくさん増えてしまい、コードが読みづらくなります。

# 関数定義
def filter_test_users(df):
    return df.filter(df.user_id != "test_user")

def normalize_score(df):
    return df.withColumn("score_norm", df.score / 100)

def add_cheater_flag(df):
    return df.withColumn("is_cheater", df.speed > 500)

# 実行（変数だらけになる）
df1 = filter_test_users(raw_df)
df2 = normalize_score(df1)
final_df = add_cheater_flag(df2)

欠点: df1, df2 といった使い捨ての変数が並び、処理の順番を入れ替えたり、一部をコメントアウトしたりするのが面倒です。

2. .transform() を使う場合（スッキリ！）

メソッドチェーンで上から下に流れるように記述できます。

# 実行
final_df = (raw_df
    .transform(filter_test_users)
    .transform(normalize_score)
    .transform(add_cheater_flag)
)

dlt.read_streamの便利さ

1. 具体的なコードと手間の違い

# 標準Sparkでの典型的な書き方
df = (spark.readStream
    .format("cloudFiles")  # Auto Loader
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/checkpoints/schema")
    .load("/input/path"))

(df.writeStream
    .format("delta")
    .option("checkpointLocation", "/checkpoints/data") # 毎回指定が必要
    .trigger(availableNow=True)
    .table("target_table"))

import dlt

@dlt.table
def target_table():
    return dlt.read_stream("/input/path") # これだけで依存関係と管理が完了

3. dlt.read_stream が圧倒的に便利な3つの理由

① 「チェックポイント地獄」からの解放

標準Sparkでは、テーブルごとに checkpointLocation を管理しなければなりません。これを間違えるとデータが重複したり壊れたりしますが、DLTはパイプラインの設定として一括管理されるため、個別のコードに書く必要がありません。

② リネージ（データフロー）の自動生成

dlt.read_stream を使うと、Databricksが自動的にテーブル間の繋がりを解析し、美しいグラフを描いてくれます。

「どのテーブルが止まっているか」「どこでレコードが減ったか」がGUIで一目瞭然になるのは、開発中のゲームのデバッグで「どのフラグが原因でバグが起きたか」を一瞬で特定できる感覚に似ています。

③ データ品質管理（Expectations）との統合

前回の質問にあった expect などの制約は、dlt.read_stream とセットで使うことで真価を発揮します。ストリーミングで流れてくるデータに対して、「ダメなデータは落とす」という処理を1行添えるだけで実装できるのはDLTならではの特権です。

Liquid Clusteringをテーブルに適用するコマンド

1. テーブルを新規作成する場合

CLUSTER BY 句を使用します。

SQL

CREATE TABLE my_catalog.my_schema.game_logs (
    event_id STRING,
    user_id STRING,
    event_time TIMESTAMP,
    event_type STRING
)
CLUSTER BY (user_id, event_type); -- ここでカラムを指定

PySpark (DataFrame API)

(df.write
    .format("delta")
    .clusterBy("user_id", "event_type") # クラスターカラムを指定
    .saveAsTable("my_catalog.my_schema.game_logs"))

マテリアライズドビューを作成するコマンド

import dlt
from pyspark.sql import functions as F

@dlt.table(
  name="stage_stats_mv",
  comment="ステージごとの集計データ（マテリアライズドビュー）"
)
def stage_stats_mv():
    return (
        dlt.read("raw_game_logs")
        .groupBy("stage_id")
        .agg(
            F.count("user_id").alias("play_count"),
            F.avg("clear_time").alias("avg_time")
        )
    )

複雑な財務処理をマテリアライズドビューを用いて、計算を効率化したい場合の具体的なコード例

import dlt
from pyspark.sql import functions as F

# 1. 通貨換算と税計算を行う中間MV（Silver層）
@dlt.table(
    name="financial_transactions_cleaned",
    comment="通貨換算と税計算済みの明細データ"
)
def transactions_cleaned():
    return (
        dlt.read("raw_sales")
        .join(dlt.read("exchange_rates"), "currency_code")
        .withColumn("amount_jpy", F.col("amount") * F.col("exchange_rate"))
        .withColumn("tax_amount", F.col("amount_jpy") * 0.10) # 簡易的な10%税計算
        .withColumn("net_amount", F.col("amount_jpy") + F.col("tax_amount"))
    )

# 2. 経営ダッシュボード用の月次集計MV（Gold層）
# ここが「計算を効率化」する本番。複雑なGROUP BYを事前に済ませて実体化する。
@dlt.table(
    name="monthly_financial_summary_mv",
    comment="月次・部署別の財務サマリー（マテリアライズドビュー）",
    table_properties={"quality": "gold"}
)
def monthly_summary():
    return (
        dlt.read("financial_transactions_cleaned")
        .groupBy(
            F.window(F.col("transaction_timestamp"), "1 month").alias("month"),
            "department_id",
            "region"
        )
        .agg(
            F.sum("amount_jpy").alias("total_sales_jpy"),
            F.sum("tax_amount").alias("total_tax_jpy"),
            F.sum("net_amount").alias("total_net_profit"),
            F.count("transaction_id").alias("transaction_count")
        )
    )

この構成のメリット（なぜ効率化されるのか）

1. 再計算の回避
   BIツール（TableauやDatabricks SQL）から参照する際、毎回「外貨換算」や「1ヶ月分の集約」を行う必要がありません。すでに計算済みの monthly_financial_summary_mv を読み込むだけなので、応答速度がミリ秒単位になります。
2. 依存関係の自動管理
   dlt.read() を使っているため、元の raw_sales（生データ）や exchange_rates（為替レート）が更新されると、DLTパイプラインが自動的にこのMVを再計算（リフレッシュ）してくれます。
3. 計算の共通化
   税計算ロジックなどを複数の部署で共有する場合、1つのMVで計算しておけば、各部署のクエリでロジックが微妙にずれるミスを防げます。

Shallow Clone vs Deep Clone

Job Parameters と Task Values

ジョブ（ワークフロー）内での「値の受け渡し」についての深い理解が問われます。

• Job Parameters:
  • ジョブ全体の開始時に渡す「定数」のようなもの。
  • dbutils.widgets.get() などで取得し、ジョブ内の全タスクで共通して利用可能。
• Task Values:
  • 「タスクAの結果をタスクBで使いたい」 という動的な受け渡しに使用。
  • dbutils.jobs.taskValues.set(key="my_key", value=123) で保存し、後続タスクで get する。
  • 試験ポイント: 「タスク間で動的なステータスや計算結果を共有する最適な方法は？」と聞かれたらこれ。

例

• Parameter Key: process_date
• Parameter Value: 2026-03-12

# パラメータの取得
process_date = dbutils.widgets.get("process_date")

# 取得した日付を使ってデータをフィルタリング
df = spark.table("raw_sales").filter(f"order_date = '{process_date}'")

Adaptive Query Execution

Spark 3.0からの目玉機能。実行中に「統計情報」を見て、クエリプランを動的に書き換える機能です。

試験で問われる「AQEができる3つの魔法」：

1. Coalescing Post-shuffle Partitions: シャッフル後に細かくなりすぎたパーティションを自動で結合し、ファイル数過多を防ぐ。
2. Switching Join Strategies: 実行中に「あれ、意外とこっちのテーブル小さいな」と判断したら、Sort Merge Joinから Broadcast Hash Join に勝手に切り替える。
3. Optimizing Skew Joins: データの偏り（スキュー）を検知して、重いパーティションを細かく分割して処理を均一化する。

is_member() 関数

• 実装例

CREATE FUNCTION sales_row_filter(region STRING)
RETURN 
  is_account_group_member('admin_group') OR region = 'Japan';

• 試験ポイント: 「特定のマネージャーには全データを、一般社員には自分の地域のデータだけを見せたい」というシナリオで、MASK ではなく ROW FILTER と組み合わせて使う。

Sort Merge

大規模テーブル同士を結合する際の「重厚な」デフォルトの結合手法です。

• ステップ:

1. Shuffle: 結合キーに基づいてデータを全ノードに再配置。
2. Sort: 各ノード内でデータをキー順に並べ替える。
3. Merge: 並んだもの同士を上からガッチャンコして結合。

• 試験ポイント: * Broadcast Hash Join が使えない（両方デカすぎる）時の最終手段。
  • Shuffleコストが高い ため、可能な限り Z-Order や Liquid Clustering でデータを整理しておくことで、このスキャンやソートを効率化できる。
  • spark.sql.shuffle.partitions の設定値が、このJoinのパフォーマンスに直結する。