Databricks 認定学習資料

テーマ切替

確認問題｜Professional ⑦ データインジェスト（配点 7%） ​

教材 07-ingestion.md の内容に基づく確認問題です（全10問）。 まず問題を解き、下の「解答・解説」で答え合わせをしてください。

問題 ​

Q1. Databricks のインジェストの 3 レイヤーと選択原則について、正しい説明はどれか。

  • A. カスタマイズ性が高い順に「マネージドコネクタ → Lakeflow パイプライン → 構造化ストリーミング」であり、最もカスタマイズ可能なマネージドコネクタから始める
  • B. 3 レイヤーはすべて独立しており、1 つのソースに対して選べるのは常に 1 レイヤーだけである
  • C. カスタマイズ性が高い順に「構造化ストリーミング（Spark API で exactly-once のフォールトトレランス、スキーマ・チェックポイントを自分で管理、Python / Scala）→ Lakeflow パイプライン（宣言的フレームワーク、オーケストレーション・監視・データ品質・エラー処理を自動管理、スキーマ・チェックポイントを自動管理、Python / SQL）→ マネージドコネクタ（Lakeflow パイプライン上に構築、ソース固有の認証・CDC・自動再試行・自動スキーマ進化を内包）」。原則は「最も管理されたレイヤーから始め、要件を満たせなければ下位（よりカスタマイズ可能な）レイヤーに降りる」
  • D. SQL でクラウドオブジェクトストレージから増分取り込みする場合、COPY INTO が最もスケーラブルで推奨される

Q2. Auto Loader の exactly-once とファイル検出モードについて、正しい説明はどれか。

  • A. 検出されたファイルのメタデータはチェックポイントの場所にあるスケーラブルなキー/値ストア（RocksDB）に保持され、これによりデータは厳密に 1 回だけ処理される。フォールトトレランスや exactly-once のために状態を自分で維持・管理する必要はない。検出モードは**ディレクトリ一覧モード（既定、LIST API でコスト増）とファイル通知モード（クラウドの通知サービス・キューを使い低コスト・低レイテンシで Databricks 推奨）**の 2 つで、どちらのモードでもファイルの検出・処理の順序は保証されない
  • B. exactly-once を実現するにはユーザーが独自の状態管理コードを実装する必要がある
  • C. 既定はファイル通知モードで、大規模時にはディレクトリ一覧モードに切り替えるとコストを削減できる
  • D. ファイル通知モードを使うと、ファイルの処理順序が到着順に保証される

Q3. Auto Loader のスキーマ推論について、正しい説明はどれか。

  • A. 初回読み取り時に対象ディレクトリの全ファイルをスキャンして推論する
  • B. 初回読み取り時、最初に検出した 50 GB または 1000 ファイル（先に達した方）をサンプリングして推論する。既定の推論型は保守的で、JSON / CSV / XML はすべて文字列（JSON の入れ子フィールド含む）、Avro / Parquet はそれぞれのスキーマでエンコードされた型。実データから型を推論させたい場合は cloudFiles.inferColumnTypes=true を指定する。スキーマ情報は cloudFiles.schemaLocation ディレクトリ内の _schemas に保存され時系列で追跡される
  • C. 既定で JSON / CSV / XML は数値・日付を含めて型推論され、cloudFiles.inferColumnTypes=false で文字列に固定する
  • D. サンプルサイズは変更できず、常に 1000 ファイル固定である

Q4. cloudFiles.schemaEvolutionMode の各モードの挙動として、正しい説明はどれか。

  • A. rescue は新列を検出するとストリームを失敗させ、再起動で更新スキーマで処理を再開する
  • B. none は新列を検出するとスキーマに追加してから失敗する
  • C. failOnNewColumns は新列をスキーマに自動追加してから失敗し、再起動で処理を再開する
  • D. addNewColumns（スキーマ未指定時の既定）＝新列をスキーマに追加後 UnknownFieldException で失敗し、再起動で更新スキーマで処理再開（既存列の型は進化しない）。addNewColumnsWithTypeWidening＝それに加えサポート型の拡大（int→long 等）、非対応の型変更はレスキュー列へ。rescue＝スキーマを進化させず失敗もせず、新列はすべて _rescued_data に記録。failOnNewColumns＝スキーマ更新または問題ファイル削除まで失敗し再起動しない（自動更新なし）。none＝進化せず新列を無視し失敗もしない。なおスキーマを指定した場合の既定は none

Q5.（シナリオ）Delta テーブルをストリーミングソースとして読んでいたが、ソーステーブルで MERGE INTO が実行されたためストリームが失敗した。対処の選択肢と使い分けとして正しいものはどれか。

  • A. ストリーミングソースの Delta は追加入力のみ受け付けるため、UPDATE / DELETE / MERGE INTO / OVERWRITE が起きるとストリームは失敗する。対処は 4 通り: skipChangeCommits（変更を無視し追加のみ処理。新規ワークロードで推奨）、完全更新、変更データフィード（CDF）（全種の変更を処理でき最も堅牢）、マテリアライズドビュー。レガシオプションの ignoreDeletes（パーティション境界の削除のみ）／ignoreChanges（DBR 12.2 LTS で skipChangeCommits に置換）もある
  • B. startingVersion と startingTimestamp を同時に指定して再起動すれば解決する
  • C. maxFilesPerTrigger を増やせば変更コミットも処理できるようになる
  • D. Delta のストリーミングソースは更新・削除も自動的に処理するため、失敗の原因は別にある

Q6. Structured Streaming のチェックポイントについて、正しい説明はどれか。

  • A. チェックポイントは複数のクエリで同一の場所を共有してよく、クエリ ID で自動的に分離される
  • B. チェックポイントディレクトリのファイルを削除しても、次回実行は前回の続きから再開される
  • C. チェックポイントディレクトリの内容は offsets（各マイクロバッチで処理するソースオフセット。中断地点から再処理なしで再開）/ commits（どのマイクロバッチがシンクにコミットされたか。exactly-once の要）/ state（ステートフルクエリの演算子・状態スキーマ・状態ストア内容のメタデータ）/ metadata（一意のクエリ ID）。クエリごとに異なるチェックポイントの場所が必須で共有してはいけない。ディレクトリのファイルを削除するか場所を変えると次回実行は新規開始になる。再起動で変えてよいのはフィルターの追加／削除・レート制限・トリガー間隔などで、入力ソースの数・種類、ステートフル操作の種類・状態スキーマ、出力シンクの種類の変更は新チェックポイントが必要
  • D. トリガー間隔を変更するには必ず新しいチェックポイントの場所が必要である

Q7. トリガーの選択について、正しい説明はどれか。

  • A. Trigger.Once が推奨で、Trigger.AvailableNow は非推奨になった
  • B. Trigger.Once は DBR 11.3 LTS 以降で非推奨であり、増分バッチには Trigger.AvailableNow（起動時点で利用可能な全データを増分バッチで処理して停止。スケジュール実行に推奨）を使う。トリガーを設定しないと processingTime=0（数ミリ秒ごと）になり、大量のストレージ API 呼び出しで予期しないクラウド課金が発生しうる。サーバーレスコンピュートでは Trigger.AvailableNow() と Trigger.Once() のみサポート（AvailableNow 推奨）で、Trigger.Continuous は Databricks 非サポート（代わりにリアルタイムモード）
  • C. トリガーを設定しない場合は 1 時間ごとに実行されるため、コスト面の心配はない
  • D. サーバーレスコンピュートでは Trigger.ProcessingTime と Trigger.Continuous がサポートされる

Q8. 出力モードとシンクの互換性について、正しい説明はどれか。

python
(streamingAggDF.writeStream
  .outputMode("update")
  .toTable("catalog.schema.agg"))   # Unity Catalog マネージド Delta テーブル
  • A. Delta シンクは append / update / complete のすべてをサポートするため、このコードは問題なく動作する
  • B. Kafka シンクは append のみをサポートし、Delta はすべてをサポートする
  • C. complete モードはデータが増えても性能が劣化しないため、ステートフル集計では常に complete を選ぶべきである
  • D. Delta Lake（Unity Catalog マネージドテーブル）は append と complete をサポートするが update は非サポートなので、このコードは動作しない。update 相当の動作は foreachBatch ＋ MERGE（ストリーミングでのマージ） で実現する。Kafka は全出力モードをサポート。また結合は append のみサポートで、重複除去は出力モードに影響されない。complete はストリーミング集計のみで動作しデータ増大で性能低下するため、多くのステートフル処理ではマテリアライズドビューの利用が推奨される

Q9. foreachBatch で Delta へ冪等に書き込むコードについて、正しい説明はどれか。

python
app_id = "orders-streaming-job"

def process_batch(batch_df, batch_id):
    if batch_df.isEmpty():
        return
    (batch_df.write.format("delta").mode("append")
        .option("txnVersion", batch_id)
        .option("txnAppId", app_id)
        .saveAsTable("catalog.schema.orders"))
  • A. txnVersion にはストリームの runId をバインドするのが正しい
  • B. batch_df.isEmpty() のチェックは不要で、空バッチは決して渡されない
  • C. txnAppId（書き込みごとに渡す一意の文字列）＋ txnVersion（batchId にバインドする単調増加の数値）により Delta Lake は重複書き込みを識別・無視するため、エラー後に同じ txnAppId・txnVersion でバッチを再実行しても重複がスキップされる。チェックポイントを削除して新チェックポイントで再起動する場合は別の txnAppId を指定する（新チェックポイントは batchId 0 から始まり、Delta は (batchId, txnAppId) を一意キーとして既知バッチをスキップするため）。空 DataFrame は渡されうる（Delta ソースの OPTIMIZE で処理対象ファイルがない、述語プッシュダウン／ファイルプルーニングで全レコード削除など）ので処理が必要
  • D. foreachBatch 内で複数のシンクへ書き込むと並列実行されるため、レイテンシは改善する

Q10. ウォーターマークと重複除去について、正しい説明はどれか。

  • A. ウォーターマークを長くすると遅延データへの耐性が下がり、レイテンシも下がる
  • B. ウォーターマークは遅延データを待つ時間の閾値であり、古い状態を自動削除してメモリエラーや遅延増大を防ぐ。しきい値内に到着したレコードは必ず処理されるが、しきい値超のレコードは処理される場合もあるが保証されない。トレードオフは「短い＝低レイテンシだが遅延データ許容度が低い」「長い＝遅延データに強いが高レイテンシ・状態増大」。ストリーム間結合の外部結合はウォーターマーク必須、複数ウォーターマークの既定はグローバルに最小値（min） を採用して遅いストリームのペースに合わせる（spark.sql.streaming.multipleWatermarkPolicy=max で変更可だが遅いストリームのデータが落ちる）。特定列の重複除去は dropDuplicates() ではなく dropDuplicatesWithinWatermark()（DBR 13.3 LTS 以降、ウォーターマーク指定が必須） または distinct() を使う
  • C. distinct() はウォーターマークなしでも状態が自動的に削除されるため安全である
  • D. 複数ウォーターマークの既定は最大値（max）で、最速のストリームを基準にする

解答・解説 ​

Q1. 正解: C Databricks の ETL スタックは「最もカスタマイズ可能なもの → 最も管理されたもの」の順に、①構造化ストリーミング（Spark API で exactly-once のエンドツーエンド・フォールトトレランス。スキーマ・チェックポイントを自分で管理。Python / Scala）、②Lakeflow パイプライン（構造化ストリーミングを拡張した宣言的フレームワーク。オーケストレーション・監視・データ品質・エラー処理を管理し、スキーマ・チェックポイントを自動管理。Python / SQL）、③マネージドコネクタ（Lakeflow パイプライン上に構築され、ソース固有の認証・CDC・エッジケース処理・API 保守・自動再試行・自動スキーマ進化を追加）の 3 レイヤー。A は順序が逆。B は誤りで、クラウドオブジェクトストレージからの取り込みは 3 レイヤーすべてで選べる。D も誤りで、SQL での増分取り込みは COPY INTO ではなく CREATE STREAMING TABLE（内部で read_files ＝ Auto Loader） がスケーラブルで堅牢なため推奨される。

Q2. 正解: A Auto Loader はファイルを検出すると、そのメタデータをチェックポイントの場所にあるスケーラブルなキー/値ストア（RocksDB）に保持し、これによりデータは厳密に 1 回だけ処理される。エラー時はチェックポイントの情報から中断地点で再開し、Delta Lake への書き込みでも exactly-once を維持する。フォールトトレランスや exactly-once のために状態を自分で維持・管理する必要はない（B が誤り）。検出モードはディレクトリ一覧モードが既定（ネイティブクラウド API の LIST で列挙。設定不要だがファイル数が多いとコスト増、継続トリガーだと特に高コスト）で、大規模時はファイル通知モード（ストレージの通知サービス／キューを自動設定してディレクトリ列挙を回避。低コスト・低レイテンシで Databricks が多くのワークロードで推奨）が推奨される（C が逆）。どちらのモードでもファイルの検出・処理の順序は保証されない（D が誤り）。

Q3. 正解: B 初回読み取り時、最初に検出した 50 GB または 1000 ファイル（先に達した方）をサンプリングして推論する。サンプルサイズは spark.databricks.cloudFiles.schemaInference.sampleSize.numBytes / .numFiles で変更できる（A・D が誤り）。既定の推論型は型不一致による進化問題を回避するため保守的で、JSON / CSV / XML はすべて文字列（JSON の入れ子フィールド含む）、Avro / Parquet はそれぞれのスキーマでエンコードされた型（C が逆）。実データから型を推論させたいなら cloudFiles.inferColumnTypes=true。スキーマ情報は cloudFiles.schemaLocation ディレクトリ内の _schemas に保存され、時系列で変化が追跡される。2 つの Parquet 間で型が違う場合は最も広い型が選ばれる（schemaHints で上書き可）。

Q4. 正解: D Auto Loader は処理中に新規列を検出すると、まず最新マイクロバッチにスキーマ推論を行い、スキーマの保存場所を新列を末尾に追加して更新し、UnknownFieldException でストリームを停止する（既存列の型は変わらない）。そのため Databricks はスキーマ変更後に自動再起動するよう Lakeflow ジョブでストリームを構成することを推奨している。5 モードは D の通りで、A・B・C はそれぞれ rescue / none / failOnNewColumns の挙動を addNewColumns の挙動と取り違えている。なおスキーマを指定した場合の既定は none であり、addNewColumns はスキーマ指定時は使えない（ただしスキーマヒントとして指定した場合は機能する）。

Q5. 正解: A ストリーミングソースとしての Delta は追加入力のみ受け付けるため、ソースで UPDATE / DELETE / MERGE INTO / OVERWRITE が起きるとストリームは失敗する（D が誤り）。対処は 4 通り: skipChangeCommits（変更を無視し追加のみ処理。新規ワークロードで推奨）、完全更新、変更データフィード（CDF）（全種の変更を処理でき最も堅牢）、マテリアライズドビュー。レガシオプションとして ignoreDeletes（パーティション境界の削除のみ）と ignoreChanges（DBR 12.2 LTS で skipChangeCommits に置換）がある。B は startingVersion と startingTimestamp の同時指定が不可である点でも誤り。C の maxFilesPerTrigger（既定 1000）は入力レート制限で無関係。

Q6. 正解: C チェックポイントと WAL（先書きログ）が連携して処理を保証する。ディレクトリの内容は offsets / commits / state / metadata（一意のクエリ ID。構成設定はオフセットログの一部として保存）。重要ルールは「クエリごとに異なるチェックポイントの場所が必要で、複数クエリで同一の場所を共有してはいけない」（A が誤り）、「チェックポイントディレクトリのファイルを削除するか新しい場所に変えると、次回実行は新規開始になる」（B が誤り）。再起動時に一般に安全な変更はフィルターの追加／削除、レート制限変更、トリガー間隔変更（D が誤り。同じチェックポイントで増分バッチ⇔時間間隔を変更できる）、mapGroupsWithState 内の UDF ロジック更新。新チェックポイントが必要（既定で不許可）なのは、入力ソースの数・種類の変更、購読 Kafka トピック／Auto Loader パスの変更、ステートフル操作の種類・状態スキーマ、出力シンクの種類。

Q7. 正解: BTrigger.Once は DBR 11.3 LTS 以降で非推奨で、増分バッチには Trigger.AvailableNow を使う（A が逆）。トリガーを設定しないと processingTime=0（数ミリ秒ごと）となり、大量のストレージ API 呼び出しで予期しないクラウド課金が発生しうるため、必ずユースケースに合ったトリガーを設定する（C が誤り）。サーバーレスコンピュートでは Trigger.AvailableNow() と Trigger.Once() のみサポート（AvailableNow 推奨）で、サーバーレスで継続的ストリーミングをしたい場合は Lakeflow パイプラインの連続（continuous）パイプラインモードを使う。Trigger.Continuous は Spark OSS の実験的機能で Databricks では非サポート（代わりにリアルタイムモード .trigger(realTime='5 minutes')、パブリックプレビュー）なので D も誤り。

Q8. 正解: D シンク別のサポートは頻出。Kafka は全出力モードをサポート（B が誤り）。Delta Lake（Unity Catalog マネージドテーブル）は append と complete をサポートするが update は非サポートなので、設問のコードは動作しない。update 相当の動作は foreachBatch ＋ MERGE で実現する（A が誤り）。また出力モードの構成が必要なのは集計を含むステートフルストリームのみで、結合は append のみサポート、重複除去は出力モードに影響されない、mapGroupsWithState / flatMapGroupsWithState は独自ロジックで出力する。complete はストリーミング集計のみで動作しデータ増大で性能低下するため、多くのステートフル処理ではマテリアライズドビューが推奨される（C が誤り）。

Q9. 正解: CforeachBatch は at-least-once しか保証しないため、Delta への冪等書き込みで exactly-once 相当を実現する。txnAppId は DataFrame 書き込みごとに渡す一意の文字列（例: StreamingQuery ID。ストリーム ID に紐付ける必要はない）、txnVersion はトランザクションバージョンとして機能する単調増加の数値で batchId にバインドする（A が誤り）。Delta Lake は txnAppId ＋ txnVersion で重複書き込みを識別・無視するため、エラーで中断後に同じ組み合わせで再実行すれば重複がスキップされる。チェックポイントを削除して新チェックポイントで再起動する場合は別の txnAppId を指定する（新チェックポイントは batchId 0 から始まるため）。空 DataFrame は渡されうるため処理が必要（B が誤り）。foreachBatch 内で複数シンクに書くと書き込みがシリアル化されレイテンシが増えるため、Databricks はシンクごとに個別の Structured Streaming ライターを使うことを推奨（D が誤り）。

Q10. 正解: B ウォーターマークはステートフルクエリで古い状態を自動削除し、メモリエラーや遅延増大を防ぐ仕組み。しきい値内に到着したレコードは必ず処理され、しきい値超のレコードは処理される場合もあるが保証されない。トレードオフは「短いウォーターマーク＝低レイテンシだが遅延データ許容度が低い」「長いウォーターマーク＝遅延データに強いが高レイテンシ・状態増大」（A が逆）。ストリーム間結合では内部結合でも各ソースにウォーターマークを推奨し、外部結合はウォーターマーク必須。複数ウォーターマークの既定はグローバルウォーターマークに最小値（min） を採用し、遅いストリームのペースに合わせて誤った遅延判定を防ぐ（D が逆）。重複除去では distinct() はウォーターマークなしだと状態が無限に増大する（C が誤り）ため、特定列の重複除去には dropDuplicatesWithinWatermark()（DBR 13.3 LTS 以降、ウォーターマーク指定が必須） を使い、全重複を保証するにはしきい値を重複イベント間の最大タイムスタンプ差より大きく設定する。