ファイル以外からも取り込む — read_files・Lakeflow Connect・JDBC と API

8.1read_files とストリーミングテーブル — ファイルリーダーの既定値、_metadata 暗黙列、VARIANT 型

SQL しか書かないチームから「landing へ毎時届く JSON を、増えた分だけテーブルに足したい」と頼まれたとします。前章の COPY INTO は先にターゲット表が要り、ファイルが数百万件規模になるとメタデータの走査が重くなります。ドキュメントの目安も、数千のオーダーなら COPY INTO、数百万以上なら Auto Loader です。かといって PySpark を書けと言えばチームは動けません。これを解くのが read_files です。

read_files はパスを表に変えるテーブル値関数

read_files はテーブル値関数、つまり SELECT 文の FROM 句にそのまま書けるファイルリーダーです。第 1 引数にパスを渡し、あとは option_key => option_value でオプションを足します。

SELECT * FROM read_files(
  '/Volumes/main/raw/orders',
  format => 'json'
);

format は省略できます。省略すると形式が自動検出され、ファイル群に対し統一されたスキーマを推論します。明示するときの値は avro、binaryFile、csv、file、json、orc、parquet、text、xml です。CSV と JSON が混在すると推論が安定しないので、本番では明示します。

PDF や画像などの非構造化ファイルは binaryFile で読みます。返るのは path、modificationTime、length、content の 4 列で、本文の自動抽出はされません。拡張子で絞るには pathGlobFilter に *.png を渡します。

FROM の直後に STREAM を付けると、同じ read_files が Auto Loader として動きます。CREATE STREAMING TABLE と組み合わせれば、SQL だけで増分取り込みが完成します。

CREATE OR REFRESH STREAMING TABLE main.bronze.orders
SCHEDULE EVERY 1 HOUR
AS SELECT *, _metadata.file_path AS src_path
   FROM STREAM read_files(
     '/Volumes/main/raw/orders',
     format => 'json'
   );

ストリーミングテーブルは、増分データ処理のサポートが加わった Delta テーブルです。ドキュメントはファイル取り込みの推奨形としてこの構文を挙げています。ただしストリーミングテーブルが動くのは Lakeflow パイプラインと、Unity Catalog を有効にした Databricks SQL の 2 か所だけです。通常のコンピュートで流しても構文が解析されるだけで、表は作られません。SCHEDULE には EVERY 1 HOUR のような間隔のほか、SCHEDULE CRON で cron 式も書けます。

STREAM を付けない read_files はただのバッチ読み取りです。過去データを一度きり表にするならこれで十分で、チェックポイントもスキーマの保存先も要りません。STREAM の有無だけが分かれ目です。

既定値を知らないと、数値の列が全部文字列になる

設定値の暗記が効くのはここです。オプション名は Auto Loader の cloudFiles.* とほぼ同じですが、既定値がそろっていないものがあります。

とくに inferColumnTypes は事故ります。Auto Loader 側は既定 false なので、JSON や CSV では入れ子フィールドを含む全列が文字列になります。「同じ JSON を読んだのに片方だけ STRING になった」の正体はこれです。useStrictGlobber も同じで、Auto Loader 側は既定 false のまま他の Spark ファイルソースと違う解釈をします。どちらも read_files が true、Auto Loader が false という向きで覚えます。

リーダー側の既定も押さえます。CSV の header は既定 false で、明示しないと列名の行がデータに混入します。sep の既定はカンマです。JSON の multiLine も既定 false で、1 行 1 レコードを前提に読むため、整形済みの JSON は全行が壊れたレコード扱いになります。その扱いを決める mode は既定 PERMISSIVE で、中身は columnNameOfCorruptRecord が指す列 (既定名 _corrupt_record) に入ります。黙って捨てる DROPMALFORMED でも、1 件で止まる FAILFAST でもありません。

型だけ直したいときは schemaHints に 'amount DECIMAL(18,2), user_info.dob DATE' のように SQL の DDL 表記で渡します。入れ子のパスも書け、指定しなかった列は推論されます。一方 schema でスキーマ全体を渡すと、その瞬間に schemaEvolutionMode が none になり新しい列は取り込まれません。型だけ直すなら schemaHints、進化を止めたいなら schema です。

_metadata は SELECT * に含まれない

どの行がどのファイル由来かを追えないと、再処理の対象を絞れません。read_files が読んだ各行には _metadata 列が付き、次の 6 つを持ちます。

• file_path — ソースファイルのフルパス (STRING)
• file_name — ファイル名 (STRING)
• file_size — バイト数 (LONG)
• file_modification_time — 最終更新時刻 (TIMESTAMP)
• file_block_start / file_block_length — ブロックの開始位置と長さ (LONG)

落とし穴は _metadata が暗黙の列で、SELECT * の結果に含まれないことです。取り込み時に拾い忘れると後から復元できません。「先週届いたファイルだけ入れ直す」という要求は必ず来るので、Bronze 表には file_path と file_modification_time だけでも入れておきます。

JSON をそのまま持つなら VARIANT

ソースの JSON が形を変え続けるとき、列に展開する設計は破綻します。持ちかたは 3 通りで、性能と柔軟性が逆を向いています。

STRUCT が最速な代わりに柔軟性は最も低い、という向きを取り違えないでください。板挟みを解くのが VARIANT 型です。型そのものと parse_json などの関数は Databricks Runtime 15.3 以降で使えますが、テーブルに VARIANT 列を持たせて読み書きするには 15.4 LTS 以降が必要で、既存の Delta 表では delta.feature.variantType-preview を有効にします。長く Public Preview で、一般提供になったのは 2026 年 7 月です。

JSON リーダーの singleVariantColumn オプションに列名を渡すと、レコード全体を 1 つの VARIANT 列として取り込めます。既定は未設定です。

SELECT payload:user.id AS user_id,
       try_variant_get(payload, '$.geo.lat', 'DOUBLE') AS lat,
       _metadata.file_path AS src_path
FROM read_files('/Volumes/main/raw/events',
                format => 'json',
                singleVariantColumn => 'payload');

取り出しかたは 3 つです。コロン演算子 :、variant_get、そして try_variant_get。最後のものはパスをたどれなくても例外にしない許容版で、崩れたソースに使います。逆に JSON 文字列から作るのが parse_json、STRUCT や MAP から作るのが to_variant_object です。VARIANT 列はパーティション列とクラスタリングキーに使えず、GROUP BY・ORDER BY・DISTINCT の対象にもできません。値の上限は 128 MiB (Runtime 17.1 以下は 16 MiB) です。

8.2Lakeflow Connect — マネージド／標準コネクタの分類と、DB CDC の4要素

Salesforce の商談データを Bronze に落としてほしい、という依頼を自前で受けると、待っているのは OAuth トークンの更新、API のレート制限、ページング、増分カーソルの保存、そして数か月おきの API 仕様変更です。取り込みロジックそのものより、ソース側の事情に振り回される時間のほうが長くなります。Lakeflow Connect は、この「ソース側の事情」を Databricks が引き受ける仕組みです。

どの層から始めるかが最初の判断

ドキュメントが示す進め方は明確です。最もマネージドな層から始め、要件を満たさない場合にだけ次の層へ降りる、というものです。逆から積み上げてはいけません。

標準コネクタは単独の層というより、Structured Streaming や Lakeflow パイプラインの中で使う部品です。取り込みロジックを細部までコードで制御したい、という要件なら答えは最下層の Structured Streaming になります。逆に、要件を確かめもせず最初から自前実装を始めるのは、この指針に真っ向から反します。

降り方の具体例を挙げます。取り込みたいソースがマネージドコネクタの一覧にない、という状況で正しい次の一手は「追加されるまで待つ」でも「手動アップロードに切り替える」でも「別の SaaS へ移行する」でもありません。標準コネクタ、コミュニティ製のコネクタ、それでも足りなければ自作のカスタムコネクタという順に、1 段ずつ下を検討します。

マネージドコネクタと標準コネクタの線引き

試験で問われるのはこの分類そのものです。マネージドは SaaS とデータベース CDC、標準はクラウドオブジェクトストレージとメッセージバスと覚えます。

Kafka は標準コネクタ側です。マネージド SaaS でもデータベース CDC でもありません。逆に SQL Server は、JDBC で読めるとしてもマネージド DB コネクタの対象で、そちらが第一選択になります。Workday も同じで、REST API を叩くノートブックを自作するのは長期の保守負荷を自分で背負う選択です。MongoDB や Cassandra のような NoSQL 製品、BigQuery や Redshift のようなデータウェアハウスは DB コネクタ (CDC) の対象ではありません。

マネージドコネクタが引き受けるのは、ソース固有の認証、変更データキャプチャ (CDC、変更ログから差分を取り出す仕組み)、細かな例外処理、API の長期メンテナンス、自動リトライ、自動スキーマ進化です。宛先は必ずストリーミングテーブルで、読み取り専用の外部表でも一時ビューでも Parquet の非管理テーブルでもありません。Unity Catalog が管理するので、通常の表と同じように GRANT で権限を配れます。

逆に、Spark の実行計画の最適化やクラスターのオートスケール設計、宛先表の列マスクや行フィルタの設計はコネクタの責務ではありません。増分の追いかたもソースごとに違い、Salesforce はカーソル列を使って前回以降の変更を拾います。

接続は Unity Catalog のオブジェクト

どのマネージドコネクタでも最初に作るのが接続 (connection) です。接続はソースの認証情報を保持する Unity Catalog のセキュラブルオブジェクト、つまりカタログや表と同じように権限を付けられる管理対象です。

ありがたみは失効のときに分かります。ノートブックの冒頭に接続文字列を直書きしたり、クラスターの環境変数にパスワードを置いたり、ジョブのタスクパラメータで毎回渡したりすると、資格情報が個々のコードへ散らばり、誰がどのソースへ届くのかを一覧できません。接続にしておけば権限付与で利用者を絞れて、監査の対象にもなります。ノートブックのシークレットスコープやクラスターのインスタンスプロファイルとは別の概念です。

SaaS は 3 要素、データベース CDC は 4 要素

SaaS コネクタの構成は、認証情報を持つ接続、サーバーレスで動く取り込みパイプライン、宛先のストリーミングテーブルの 3 つです。ここにゲートウェイもステージングも出てきません。データベース CDC コネクタはこれに 2 つが加わり、次の 4 要素になります。設問では部品を 1 つ入れ替えた選択肢が並ぶので、SaaS にゲートウェイとステージングは付かないという切り分けを先に思い出します。

コンピュートの割り当ては入れ替えて出題されます。ゲートウェイがクラシック、取り込みパイプラインがサーバーレスです。SaaS ソースは完全にサーバーレス専用、データベースソースはクラシックとサーバーレスの両方で動かせますが、ゲートウェイだけはクラシックを必要とするため、クラシックコンピュートをサポートしないワークスペースにはデプロイできません。オンプレミスの SQL Server が相手なら、そこへ疎通できる経路の用意が前提条件になります。

ゲートウェイを止められない理由も設問になります。ソース側で変更ログが切り詰められる前に変更を捕捉しなければならないからです。宛先を最適化するためでも、サーバーレスの起動を速くするためでも、権限を再評価するためでもありません。副作用として、取り込みパイプラインがアイドルの間もゲートウェイのクラシックコンピュートはプロビジョニングされ、課金され続けます。マネージドコネクタはコンピュートベースの価格モデルで、データベースソースではクラシックとサーバーレス双方の DBU (Databricks のコンピュート課金単位) が発生しうる、と理解しておきます。転送レコード数による課金でも、ストレージだけの課金でもありません。

ステージングストレージは、ゲートウェイのデプロイ時に Unity Catalog ボリュームとして自動作成されます。置き場所のカタログとスキーマは選べて、置かれたデータは 30 日後に自動削除されます。ソース DB 側の一時テーブルでも、DBFS ルート配下の固定パスでも、宛先 Delta 表への相乗りでもありません。長期間成功していないパイプラインを放置すると宛先に欠落が生まれるので、復旧にはフルリフレッシュが要ります。

既定の挙動を 2 つ覚える

1 つ目はスキーマ進化です。明示的にオプトアウトしない限り、新しい列は次の実行で自動的に取り込まれ、その列が現れる前の行は空のままになります。ソースで削除された列は宛先から物理削除されず、テーブルプロパティで非アクティブとして扱われます。列が増減しても取り込みが止まったり、表を作り直したりする必要はありません。この自動処理から外したいときは、取り込む列を明示的に列挙するか、UI で将来の列を無効にします。

2 つ目は実行が重なったときです。スケジュールされた更新 N がまだ走っている状態で次の時刻が来ると、更新 N+1 はスキップされ、更新 N+2 から再開されます。並列に走ることも、パイプライン全体がエラーで止まることも、実行中の更新が強制終了されることもありません。間隔を詰めすぎると実質的な更新頻度が想定の半分になる、という運用上の含意があります。

マネージドコネクタが解決しないことも押さえます。取り込むのは Bronze 相当の生データまでで、Silver や Gold の集計まで自動生成されるわけではありません。レイテンシもソースとモードによって秒から分まで幅があり、一律の保証はありません。コネクタは読み取り専用で、ソース DB を書き換えることもありません。列名の不一致や型エラーが残るときは、ソース層で救済列を有効にして値を _rescued_data に残し、パイプライン層では Expectations (パイプラインに宣言するデータ品質ルール) で ON VIOLATION DROP ROW のように違反行の扱いを決めます。エラーのたびに FAIL UPDATE で全体を止めるのは最後の砦であって、常用する設定ではありません。

選び損ねると何が起きるかも見ておきます。Salesforce の取引データを Auto Loader で取ろうとすると、まず「CSV をエクスポートしてストレージへ置く」工程を誰かが作ることになります。その工程が壊れれば取り込みも止まり、どこまで取ったかの判定も自前です。組み込みの SaaS コネクタを使えば、この工程ごと消えます。オンプレミスの SQL Server を毎晩 JDBC でフルテーブル読み直す設計も同じ種類の間違いで、夜間の負荷は表が育つほど増え、そもそも増分になりません。

「取り込む」のか「連合する」のか

外部データベースを Databricks から扱う道は 2 系統あります。Lakeflow Connect でコピーを作るのが取り込み、Lakehouse Federation でコピーを作らずに参照するのが連合です。どちらも Unity Catalog の接続で外部 DB につなぐため、混同したまま設問に入ると必ず外します。

判断軸は「そのデータを繰り返し重く読むか」です。毎日何度も参照して結合する相手なら取り込み、たまに数千行を確認するだけなら連合が向きます。設問に「オンプレミスの SQL Server の変更を継続的に取り込み、宛先へ増分適用したい」とあれば、答えはゲートウェイを伴うマネージド DB コネクタです。ここで外部表として参照する選択肢を選ぶと、要件の「取り込み」を「参照」に読み替えたことになります。連合側の設定手順と制約は本章では扱わず、別の解説へ譲ります。

8.3ファイル以外から取り込む — JDBC のチューニング、Statement Execution API、ボリュームへのアップロード

マネージドコネクタの一覧にない RDB を、今夜中に 1 テーブルだけ読みたい。現実解は JDBC です。ところが素直に書くとたいてい遅く、遅さは 3 つに分解できます。原因を取り違えると何をいじっても速くなりません。

JDBC の基本形は 4 つのオプション

ドキュメントが例に挙げる構成は、接続文字列の url、対象テーブルの dbtable、認証用の user と password の 4 つです。cloudFiles.* でも warehouse_id でもありません。スキーマは指定不要で、Spark がデータベース側からスキーマを読み取り、Spark SQL の型へマッピングします。

df = (spark.read.format("jdbc")
      .option("url", "jdbc:postgresql://db.internal:5432/sales")
      .option("dbtable", "public.orders")
      .option("user", user)
      .option("password", password)
      .load())

厳密に必須なのは url と、dbtable または query のどちらか一方です。資格情報は平文で書かず、シークレット管理に置くことが推奨されています。

遅さの原因を 3 つに分ける

1 つ目は往復の多さです。fetchSize は 1 回のやり取りで取得する行数を決めます。既定はドライバーによって違い、Oracle は 10 です。既定の小さいドライバーでは 100 以上へ上げるだけで大きく改善し、多くのデータセットでは数千まで効きます。上げすぎるとメモリ不足になります。

2 つ目は並列度です。既定では 1 本の接続で順に読むため、大きなテーブルではここが天井です。並列化には 4 つをセットで指定します。値が一様に分布したインデックス付きの列を partitionColumn に、範囲を lowerBound と upperBound に、分割数を numPartitions に渡します。偏った列だと 1 つのパーティションだけ巨大になります。

注意が要るのが numPartitions の大きさです。小規模クラスターではエグゼキューターのコア数と同じ値が勧められ、コード例は 8 です。そのうえで 「この値を 50 より上に設定するのは要注意」と警告しています。数百まで上げると同時クエリでリモート DB を圧迫し、かえって遅くなります。

3 つ目は転送量です。全件読んで Spark 側で filter しても流れる量は減りません。dbtable に別名付きサブクエリを書き、絞り込みをソース側で実行させます。

.option("dbtable", "(select id, amount from orders where status = 'PAID') as t")

REST から SQL を投げる — Statement Execution API

社内の Web アプリから集計結果を取りたいだけなのに、JDBC / ODBC ドライバーを配るのは重すぎます。この用途が SQL Statement Execution API です。

POST /api/2.0/sql/statements に warehouse_id と statement を渡すと実行され、状態と結果は GET /api/2.0/sql/statements/{statement_id} で取得します。ジョブの /api/2.0/jobs/run-now に SQL 文を渡すのではありません。

wait_timeout に 0s を指定すると結果を待たずに文の ID と状態が返り、あとからポーリングする非同期になります。既定の 10 秒で終わらなければ statement_id と状態だけが返ります。既定の INLINE は応答本体に結果を埋め込み、25 MiB を超えると失敗ステータスが返って文は取り消されます。それ以上は EXTERNAL_LINKS を指定し、短命な署名付き URL からダウンロードします。この URL には Authorization ヘッダーを付けません。

文に値を埋め込むときは :parameter_name のプレースホルダを書き、parameters に名前と値を並べます。大きな結果はチャンクに分かれるので順にたどります。外部アプリからの問い合わせ口であり、TB 級を流し込む経路ではありません。

手元のファイルを一度だけ持ち込む

200 MB の CSV が 1 本、今回きり。パイプラインを組むのは過剰で、ローカル PC を JDBC で参照させる構成は成立しません。Catalog Explorer の Add or upload data から、Unity Catalog ボリュームへ直接置けます。UI 経由のアップロードは 1 ファイル 5 GB までで、超えるなら Databricks SDK for Python を使います。

必要な権限は 3 つで、対象ボリュームの WRITE VOLUME、親スキーマの USE SCHEMA、親カタログの USE CATALOG です。カタログとスキーマを忘れて「書けるはずなのに見えない」と詰まりがちです。置いたあとは /Volumes/カタログ/スキーマ/ボリューム/... を read_files に渡せば、この章の最初に戻ってきます。定常運用で手作業のアップロードを続けるのはアンチパターンです。

方式の早見表

この表の右列がそのまま設問の分かれ目です。経路は覚えやすい一方、数字を落とすと 2 択で止まります。