取り込み時にスキーマをどう扱うか — Auto Loader のスキーマ進化と COPY INTO

7.1スキーマ強制と進化 — mergeSchema / autoMerge.enabled / overwriteSchema / ALTER TABLE ADD COLUMNS の使い分け

ソース側の担当者が JSON にフィールドを一つ足す。それだけで夜間バッチが朝には止まっている、あるいは黙って列が捨てられる。どちらが起きるかは書き込み先の設定で決まります。Delta Lake が既定で選んでいるのは「止まる方」で、これをスキーマ強制 (schema enforcement) と呼びます。この節では書き込み側の規則だけを片付けます。ファイルを読む側の話は次の節からです。

既定の挙動をコードで確認する

列が id と name の二つしかない Delta 表に、amount を足した DataFrame を追記してみます。

df_new = spark.createDataFrame(
    [(1, 'A', 100), (2, 'B', 200)],
    ['id', 'name', 'amount'])

df_new.write.format('delta') \
    .mode('append') \
    .save('/Volumes/main/raw/sales')

結果は AnalysisException です。amount が自動で追加されることはなく、amount が黙って捨てられることもありません。壊れた形のデータを下流へ流さないことを優先した設計で、試験では「mergeSchema を指定していない append の結果はどうなるか」という形で繰り返し問われます。答えは常に書き込みが失敗するです。「追加列は警告なしに無視される」「新列が自動で追加される」「テーブルが上書きされる」はすべて誤りの選択肢として並びます。

進化を許可する 4 つの手段

強制を恒久的に切るのではなく、この書き込みに限って進化を許すと宣言するのが Delta の作法です。手段は四つあり、効く範囲と危険度が違います。

ここで問われる判断は「安全に新しい列を足す方法を選べ」という形です。正解になるのは mergeSchema と ALTER TABLE ... ADD COLUMNS の二つで、overwriteSchema は表定義ごと差し替える破壊的な操作なので「安全な追加」の答えにはなりません。ALTER TABLE で追加した列は nullability が既定で true、つまり NULL を許す列になり、既存行の値は NULL のままです。

ALTER TABLE bronze.sales
  ADD COLUMNS (amount DECIMAL(18,2) COMMENT '税抜金額' AFTER name);

ここまでの四つはどれも列を足す方向にしか働きません。mergeSchema で追加された列は表スキーマの末尾に付きます。列の改名と削除をしたい場合は列マッピング (物理ファイル上の列名と論理的な列名を切り離す機能) を有効にしたうえで ALTER TABLE ... RENAME COLUMN と ALTER TABLE ... DROP COLUMN を使います。しかも DROP COLUMN はメタデータ上で列を消すだけで、ファイルの中のデータは残ります。物理的に消すには REORG TABLE のあとに VACUUM が要ります。セッション設定の autoMerge.enabled が本番向きでないのも同じ理由で、列名を打ち間違えたまま書き込むと、そのタイプミスがそのまま表定義として定着してしまいます。

MERGE と INSERT には専用の句があり、対応バージョンが別々なので数字だけ押さえます。SQL の MERGE WITH SCHEMA EVOLUTION 句は Databricks Runtime 15.2 以降、Python と Scala のテーブル API で .withSchemaEvolution() を挟む書き方は Databricks Runtime 15.4 LTS 以降、INSERT WITH SCHEMA EVOLUTION は Databricks Runtime 18.1 以降です。ここを取り違える設問が出ます。

スキーマ推論・スキーマ強制・スキーマ進化を混ぜない

この三つは名前が似ていますが、働く場所も、失敗したときに起きることも別です。混同したまま設問を読むと、選択肢が全部それらしく見えてしまいます。

切り分けは症状で付きます。エラーメッセージに列名と型の対応表が出て書き込みが止まっているなら強制側、エラーは出ないのに全列が STRING になっているなら推論側です。読み取り側が新しい列を認識できていなければ、書き込み側で mergeSchema をいくら立てても列は増えません。スキーマ事故は、この二層のどちらで詰まっているかを決めるところから始まります。

強制をどの層でかけるか

メダリオンアーキテクチャ (Bronze・Silver・Gold の三層でデータを磨いていく設計) では、強制をかける場所も決まっています。Bronze はソースの生データをロスレスに残す層なので、パースと最小限の型付けだけにとどめ、検証は緩くします。正規化・重複排除・スキーマ強制と品質ルール (Expectations) は Silver 層に置きます。Bronze で厳格に弾くと、弾かれたデータが手元に残らず再処理ができなくなります。Delta 側のスキーマ進化そのもの、型の拡張、列マッピング、制約の詳細は当サイトの「Delta スキーマ進化」「Delta 制約」の記事で扱っているので、深掘りはそちらに送ります。

7.2Auto Loader (2) — スキーマ推論の上限（50GB / 1000ファイル）、schemaEvolutionMode 4値、rescuedDataColumn、cleanSource

Auto Loader を本番に載せたチームがまず驚くのは、金額のはずの列がすべて文字列になっていること、次に、上流が列を一つ足した翌朝にストリームが止まっていることです。どちらも設定を間違えたわけではなく、既定値どおりの動作です。この節では、その既定値を数値と値の名前で覚えます。試験がこの領域から出す問題は、ほとんどが「オプション名」「既定値」「モードの挙動」の三つに集約されます。

推論の材料は最初の 50 GB か 1000 ファイルしかない

Auto Loader は入力ディレクトリの全ファイルを読んでスキーマを決めているわけではありません。最初に検出した 50 GB か 1000 ファイルのうち、先に上限へ達した方までをサンプリングして推論します。1 億ファイルあるバケットでも、見るのは先頭のごく一部です。上限は次の二つの Spark 設定で変えられます。

この上限があるせいで、サンプルに現れなかった列や、後から届いたファイルにしかない型は推論から漏れます。漏れた分をどう扱うかが、このあとの進化モードと救済列の役割です。

推論しても型がつかない — inferColumnTypes

JSON・CSV・XML のように型情報を持たない形式では、Auto Loader は入れ子フィールドを含む全列を文字列として推論します。「数値のはずの列が全部 STRING になった」という設問はこの仕様を突いたもので、原因は cloudFiles.inferColumnTypes が既定 false であることです。true にするとサンプルデータから具体的な型を推論します。schemaLocation を指定し忘れたせいではありませんし、救済列が型を奪っているわけでもありません。

推論結果の置き場所 — schemaLocation

スキーマ推論と進化を使うには cloudFiles.schemaLocation が必須です。指定したパスに推論したスキーマとその変遷が保存され、これを省くとストリーム再起動のたびに推論をやり直すことになり、進化モードも成立しません。

df = (spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .option("cloudFiles.schemaLocation", "/Volumes/main/ops/_schemas/events")
      .option("cloudFiles.inferColumnTypes", "true")
      .option("cloudFiles.schemaHints", "amount DOUBLE, user_info.dob DATE")
      .option("cloudFiles.rescuedDataColumn", "_rescue")
      .load("/Volumes/main/raw/events"))

schemaEvolutionMode の 5 つの値

新しい列が現れたときに何をするかを決めるのが cloudFiles.schemaEvolutionMode です。長く 4 値と説明されてきましたが、型の拡張を伴うモードが加わって現在は 5 値あります。

引っかかりやすいのが addNewColumns と failOnNewColumns の区別です。どちらもストリームは落ちます。違うのは再起動したときで、addNewColumns は新列を取り込んで続きから走り、failOnNewColumns は人手でスキーマを直すまで走りません。Lakeflow Jobs (Databricks のジョブ実行基盤) のタスク再試行と addNewColumns を組み合わせると、人が介在せずに列追加を吸収できます。PERMISSIVE は後述するパーサーのモード名であって schemaEvolutionMode の値ではなく、誤答の常連です。

扱う範囲にも限りがあります。schemaEvolutionMode が面倒を見るのは列の追加と型の拡張までで、ソース側で列が削除されたことを検知して表から消す機能ではありません。また、ディレクトリ構造から取り出す Hive 形式のパーティション列は進化の対象外で、増やしたいときは cloudFiles.partitionColumns にカンマ区切りで明示します。

自分でスキーマを渡すと既定値が 3 か所裏返る

ここまでは推論に任せる前提でした。.schema(...) で自分のスキーマを渡すと、既定値が同時に三つ変わります。

つまり固定スキーマにして何も足さないと、新しい列は進化もせず救済もされず静かに消えます。「スキーマを固定したらデータが欠けた」という障害報告の大半はこの組み合わせで、不具合ではありません。固定したまま取りこぼしを見えるようにするなら、rescuedDataColumn を明示するか、モードを rescue か failOnNewColumns に切り替えます。

救済列 — rescuedDataColumn

スキーマに収まらなかった値を捨てないための逃がし場所が救済列です。既定の列名は _rescued_data で、cloudFiles.rescuedDataColumn に別名を渡せば変えられます。中身は救済された列と、そのレコードのソースファイルパスを含む JSON です。対象になるのは、推論または宣言したスキーマに存在しない列、型が合わずキャストできなかった値、大文字小文字だけが違う列名の三種類です。

破損レコードの扱いとは別物なので混ぜないでください。JSON や CSV のリーダーには mode オプションがあり、既定は PERMISSIVE、壊れたレコードは columnNameOfCorruptRecord で指定した列 (既定名 _corrupt_record) に入ります。DROPMALFORMED で捨てられたり FAILFAST で落ちたりするのは、JSON や CSV として本当に壊れているレコードだけで、単なる型の食い違いはこちらではなく救済列に回ります。「想定外のカラムに値が入る」という現象の正体はたいてい救済列で、SELECT _rescued_data FROM ... で中身を確認できます。

schemaHints で型だけを直す

推論が STRING や DOUBLE に倒れて困る列を名指しで直すのが cloudFiles.schemaHints です。値は JSON ではなく SQL の DDL 表記で、"amount DOUBLE, tags map<string,string>" のようにカンマ区切りで並べ、user_info.dob DATE のような入れ子のパスも指定できます。指定しなかった列は通常どおり推論されます。ヒントを書いたのに型が変わらないときは、列名の綴り違いか別名で書いているのが原因なので、正しい「列名 型」で書き直し、schemaLocation を保ったまま再起動するのが定石です。チェックポイントを消してやり直すのは、再処理コストと重複を招くだけです。

処理済みファイルの後始末 — cleanSource

取り込みが終わったファイルを入力ディレクトリに置きっぱなしにすると、一覧取得のコストが延々と増えます。cloudFiles.cleanSource はこれを自動で片付けるオプションで、Databricks Runtime 16.4 以降で使えます。

MOVE の制約は言い換えると一つです。バケットやコンテナをまたぐ移動はサポートされません。

まとめて覚える既定値

取り込み状況の確認には cloud_files_state() テーブル関数を使います。チェックポイントのパスか TABLE(ストリーミングテーブル名) を渡すと path・size・discovery_time・commit_time・ingestion_state などの列が返り、commit_time が NULL のファイルはまだ処理されていないことを意味します。ingestion_state には INGESTED や PROCESSING、SKIPPED_CORRUPTED といった値が入ります。バックログの大きさは Streaming Query Listener が報告する numFilesOutstanding と numBytesOutstanding で測ります。

7.3COPY INTO 完全解剖 — 冪等性の記録先、FORMAT_OPTIONS と COPY_OPTIONS、VALIDATE / BY POSITION / FILES と PATTERN / WITH CREDENTIAL

SQL しか書かないチームが最初に困るのは、たいてい同じ場面です。ジョブが落ちたので手で流し直したら同じ行が二重に入った。毎晩ディレクトリをまるごと指定したいが、昨日のファイルまで読み直されるのが怖い。COPY INTO は、この「同じものを二度入れたくない」を SQL 一文で解決するコマンドです。Auto Loader がストリーミング側の道具なのに対して、COPY INTO はバッチの世界で冪等性を担保する道具です。

構文の全体像

COPY INTO target_table [ BY POSITION | ( col_name [, col_name ...] ) ]
FROM { source_clause | ( SELECT expression_list FROM source_clause ) }
FILEFORMAT = data_source
[ VALIDATE [ ALL | num_rows ROWS ] ]
[ FILES = ( file_name [, ...] ) | PATTERN = glob_pattern ]
[ FORMAT_OPTIONS ( { data_source_reader_option = value } [, ...] ) ]
[ COPY_OPTIONS ( { copy_option = value } [, ...] ) ]

source_clause
  source [ WITH ( [ CREDENTIAL ... ] [ ENCRYPTION ... ] ) ]

FILEFORMAT は公式ドキュメントの中で記述が食い違っているので両方覚えます。SQL 構文リファレンスが列挙しているのは CSV / JSON / AVRO / ORC / PARQUET / TEXT / BINARYFILE の 7 種類ですが、COPY INTO の概要ページは XML を含めた 8 種類を挙げ、XML のページには FILEFORMAT = XML の実例が載っています。XML は Databricks Runtime 14.3 以降で、通常は FORMAT_OPTIONS ('rowTag' = '...') を併用します。左辺がターゲット表、右辺がソース URI で、これを逆に書いた選択肢が誤答として並びます。

ターゲット表は先に無いと動かない

COPY INTO は既存の Delta 表しか受け付けません。存在しない表名を書いても表は自動作成されず、一時ビューにも外部表にもなりません。では取り込むデータの列構成が未確定のときはどうするか。ドキュメントが示す手順は、列を持たない空のプレースホルダ Delta 表を先に作り、mergeSchema を true にして推論させるというものです。スキーマを省略できるのは Databricks Runtime 11.3 LTS 以降で、スキーマ進化に対応した形式に限られます。

CREATE TABLE IF NOT EXISTS my_pipe_data;

COPY INTO my_pipe_data
FROM 's3://my-bucket/pipeData'
FILEFORMAT = CSV
FORMAT_OPTIONS ('mergeSchema' = 'true', 'delimiter' = '|', 'header' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');

この空表は COPY INTO 専用です。ドキュメントは、スキーマの無い Delta 表への書き込みに INSERT INTO と MERGE INTO はサポートされないと明記しています。データが入ってはじめて普通の表として問い合わせられます。権限面では、アカウント管理者がクラウドストレージへのアクセスを構成したうえで、実行するユーザーに対象のカタログ・スキーマ・ボリュームに対する CREATE TABLE 権限が必要です。

冪等性はどこに記録されているか

COPY INTO の最大の性質は再実行しても重複しないことです。仕組みは単純で、ソースの場所にあるファイルのうち既に読み込んだものは次回以降スキップされるだけです。記録はターゲット表側に置かれ、COPY INTO が作ったメタデータファイルは Databricks Runtime 15.2 以降なら VACUUM で掃除できます。ファイル内容のハッシュを毎回計算しているわけでも、内部で MERGE に切り替わっているわけでもありません。帰結が三つ出ます。

• 同じ COPY INTO を二度実行すると、二度目は何も取り込まれず、例外も起きない。
• ロード後にソース側でファイルの中身が書き換えられていても、取り込み済みとして扱われスキップされる。差分追記も行更新も起きず、チェックサム不一致で失敗することもない。
• ファイル名を毎回変える運用にすると別ファイル扱いになり、冪等スキップが効かなくなる。

この冪等性を意図的に外すのが COPY_OPTIONS ('force' = 'true') です。既定は false で、true にすると取り込み済みかどうかにかかわらず全ファイルが読み込まれます。書き換え後の内容を反映したいときやバックフィルには必要ですが、本番ジョブに付けっぱなしにすると重複が積み上がります。

運用の側から冪等性を壊す例も覚えます。毎回 DROP TABLE と CREATE TABLE で作り直す、事前に TRUNCATE TABLE で空にする、ファイル名に毎回タイムスタンプを付ける。どれも記録を消すか鍵を変えるので全件再取り込みになります。冪等性はファイルのパスを鍵にして成立していると押さえてください。

同じ「二度取り込まない」でも、Auto Loader とは記録場所が違います。COPY INTO はターゲット表側、Auto Loader はチェックポイント内のキーバリューストア (RocksDB) です。記録が別々である以上、同じ表に両方を常用すると互いの履歴を参照せず二重取り込みの温床になります。ただしドキュメントは、Auto Loader のストリームを走らせたまま COPY INTO で一部のファイルだけ入れ直す使い方は認めています。

FORMAT_OPTIONS と COPY_OPTIONS を取り違えない

この振り分けは単独で一問になるほど頻出です。

COPY_OPTIONS に二つしか無いと知っていれば、COPY_OPTIONS ('header' = 'true') や COPY_OPTIONS ('multiLine' = 'true') を並べた選択肢は読んだ瞬間に消せます。「破損ファイルを飛ばして続行したい」は FORMAT_OPTIONS ('ignoreCorruptFiles' = 'true')、「複数行にまたがる整形済み JSON を読みたい」は FORMAT_OPTIONS ('multiLine' = 'true') です。CSV の header も JSON の multiLine も既定は false で、明示しないと先頭行がデータ行になります。

やっかいなのは mergeSchema が両方の句に書けることです。FORMAT_OPTIONS 側は複数ファイル間でスキーマを突き合わせる読み取り時の指定、COPY_OPTIONS 側はターゲット表のスキーマを進化させる指定です。列が増えたときに表を追随させたいなら COPY_OPTIONS 側。なお Auto Loader の救済列は COPY INTO では使えないので、取りこぼしたくないなら mergeSchema で表側を広げるしかありません。

VALIDATE — 書かずに確かめる

本番表へ書く前に確かめたいときに使うのが VALIDATE 句です。Databricks Runtime 10.4 LTS 以降で使え、付けるとデータは検証されるだけでテーブルには一切書き込まれません。検証されるのはパース可否、表のスキーマと一致するか (進化が必要か)、nullability と CHECK 制約を満たすか、の三点です。

• VALIDATE ALL: 全データを検証する。VALIDATE を書いたときの既定。
• VALIDATE 100 ROWS: 指定した行数を検証する。50 未満の数を指定すると 50 行以下のデータのプレビューが返る。

「書き込む前に確かめたい」に対して一時テーブルへ COPY INTO する案は、実際に書き込みが起きるので不正解です。

BY POSITION — 列名が合わないとき

ソースの列名がターゲット表と一致しないが並び順は一致している。この状況を解決するのが BY POSITION で、ソース列とターゲット列を序数位置で対応付け、型のキャストも自動で行います。使えるのはヘッダーなし CSV だけで、FILEFORMAT = CSV の指定と FORMAT_OPTIONS ("headers" = "false") が必要です (これが既定値でもあります)。

COPY INTO bronze.sales BY POSITION
FROM 's3://my-bucket/headerless/'
FILEFORMAT = CSV;

表記に注意します。BY POSITION で書くのは複数形の headers で、CSV リーダーの header とは別のキーです。誤答肢に並ぶ FORMAT_OPTIONS ('header' = 'false') は先頭行をデータとして読む指定でしかなく、列名の食い違いは解決しません。対応付けではターゲット表の IDENTITY 列 (連番が自動採番される列) と GENERATED 列 (式から自動計算される列) が無視され、残った列数とソース列数が合わなければエラーになります。

FILES と PATTERN — 対象を絞る二つの句

この二つは同時に指定できません。FILES で明示したファイルでも冪等性の判定は生きていて、取り込み済みなら黙ってスキップされます。PATTERN は読み込む対象を選ぶ句であり、除外を書く句でも履歴管理を無効化する句でもありません。

WITH ( CREDENTIAL ... ) とロード時の変換

ソースパスへのアクセス資格情報は、FROM 句のパスに続けて WITH で渡します。COPY_OPTIONS や FORMAT_OPTIONS に credential というオプションは存在しません。名前付き資格情報のほかに一時資格情報 (AWS なら AWS_ACCESS_KEY / AWS_SECRET_KEY / AWS_SESSION_TOKEN、Azure なら AZURE_SAS_TOKEN) も渡せ、同じ括弧に ENCRYPTION も並べられます。

COPY INTO bronze.sales
FROM 's3://my-bucket/raw/' WITH ( CREDENTIAL my_named_credential )
FILEFORMAT = PARQUET;

ロードしながら型を変えたり列を落としたりしたい場合は、FROM 句にサブクエリを書きます。オプション句は key = value しか受け取らないので、式を書ける場所はここだけです。

COPY INTO my_delta_table
FROM (SELECT to_date(dt) dt, event as measurement, quantity::double
      FROM 's3://my-bucket/avroData')
FILEFORMAT = AVRO;

同時実行と選定基準

同じ表に対する複数の COPY INTO は、それぞれが扱う入力ファイルの集合が互いに素 (共通のファイルを一つも含まない状態) であれば、いずれも最終的に成功します。重なるとトランザクション競合になります。ただし 1 つのコマンドに複数ファイルを渡す方が速いため、性能目的の並行実行は推奨されていません。

選定の目安は明快です。時間をかけて取り込むファイルが数千のオーダーなら COPY INTO、数百万以上が見込まれるなら Auto Loader。スキーマが頻繁に変わる場合も Auto Loader が向きます。SQL だけでより拡張性の高い形を求めるなら、ドキュメントが示す代替は CREATE STREAMING TABLE と read_files によるストリーミングテーブルです。

三面比較 — 同じ「新しい列」に三者はどう答えるか

並べると、救済列を持つのは Auto Loader だけだと分かります。優劣ではなく、必要なのが救済かチェックポイント不要かで選びます。