目次 › 第2章 › 2-5

第2章 データ取り込みとロード / 想定学習時間:30〜40分 / 最終確認:2026年8月

2-5. JDBC/ODBC・RESTクライアントによる取り込み

1. 自前実装パターンの位置づけ

2-4 で見たとおり、対応するマネージドコネクタやパートナーコネクタがあるならそちらが優先です。しかし現実には、コネクタが対応していないデータベースや社内 API、特殊な抽出条件が必要な場面が残ります。そこで登場するのが、ノートブックに抽出コードを書き、Lakeflow Jobs(旧 Workflows)でスケジュールする自前実装パターンです。

2. JDBC による外部データベースの読み取り

Spark には JDBC データソースが組み込まれており、spark.read.format("jdbc") で外部データベースのテーブルを DataFrame として読み取れます。

# 認証情報は Databricks シークレットから取得(コードに直書きしない)
user     = dbutils.secrets.get(scope="ingest", key="db_user")
password = dbutils.secrets.get(scope="ingest", key="db_password")

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

# UC 管理 Delta テーブルへ書き込み
df.write.mode("append").saveAsTable("main.bronze.orders")

いったんクラウドストレージに中間ファイルとして書き出してから 2-2・2-3 の仕組みでロードする構成も可能ですが、DataFrame として読めているなら直接 UC 管理テーブルへ書き込むのが最短です。なお ODBC は主に BI ツールなど外部クライアントから Databricks へ接続する文脈で登場する規格で、ノートブックからの読み取りでは JDBC を使うのが通常です。

3. 増分抽出:watermark 列

JDBC 読み取りには COPY INTO や Auto Loader のような「既処理の自動記録」がありません。毎回全件を読むとソース DB に負荷がかかるため、watermark 列(更新日時や連番 ID など、単調に増える列)を使って「前回読んだところから先だけ」を抽出します。

# 前回ロード済みの最大更新日時をターゲットテーブルから取得
last_ts = (spark.table("main.bronze.orders")
           .selectExpr("max(updated_at)").first()[0])

query = f"(SELECT * FROM public.orders WHERE updated_at > '{last_ts}') AS t"

df_inc = (spark.read.format("jdbc")
  .option("url", jdbc_url)
  .option("dbtable", query)
  .option("user", user)
  .option("password", password)
  .load()
)
df_inc.write.mode("append").saveAsTable("main.bronze.orders")

4. REST API クライアントによる取得

社内 API や外部サービスの REST API からの取り込みは、Python の requests などの HTTP クライアントで取得し、DataFrame に変換して Delta へ書き込みます。

import requests

token = dbutils.secrets.get(scope="ingest", key="api_token")

resp = requests.get(
    "https://api.example.internal/v1/products",
    headers={"Authorization": f"Bearer {token}"},
    timeout=30,
)
resp.raise_for_status()
records = resp.json()["items"]   # 辞書のリスト

df = spark.createDataFrame(records)
df.write.mode("overwrite").saveAsTable("main.bronze.products")

REST 取得のコードはドライバノード上の単一プロセスで動く点に注意してください。Spark の分散処理が効くのは DataFrame 化した後です。巨大なデータを1つの API から吸い出す用途には向かず、その場合はソース側にエクスポート機能(ストレージへのファイル出力)がないかを先に検討します。

5. 運用の3点セット

5-1. 認証情報はシークレットで管理する

パスワードや API トークンをノートブックに直書きするのは厳禁です。Databricks シークレット(dbutils.secrets.get(scope, key))に格納し、コードからは参照だけを行います。シークレットの値はノートブックの出力上では自動的に伏せ字(REDACTED)になります。

5-2. ドライバの制約

主要なデータベースの JDBC ドライバは Databricks Runtime に同梱されていますが、それ以外のデータベースではドライバの JAR をクラスタにライブラリとしてインストールする必要があります。また、サーバーレスなど環境によってはライブラリ導入やネットワーク到達性に制約があるため、接続先への経路(プライベートネットワーク等)も含めて事前確認が必要です。

5-3. Lakeflow Jobs でスケジュールする

抽出ノートブックは手動実行で終わらせず、Lakeflow Jobs のタスクとして登録して定期実行します。リトライ・タイムアウト・失敗時の通知を Jobs 側に設定すれば、自前実装でも運用品質を確保できます(第4章で詳述)。

練習問題