Databricksで押さえるSchema Registryパターン: Avro/JSON・Auto Loader・スキーマ進化

ストリーミングやバッチ取り込みにおいて、上流のスキーマ変更はパイプライン障害の主要因です。 Databricksでは「Confluent Schema Registryとの連携」「Auto Loaderのスキーマ推論・進化」「Delta Lakeのスキーマ進化」を組み合わせてスキーマの変更を安全に吸収します。 本稿ではこれらのパターンを実務と試験の両面から整理します。

スキーマレジストリとは何か

スキーマレジストリは、メッセージのスキーマ定義（フィールド名・型・デフォルト値）を一元管理するサービスです。 Kafka等のメッセージングシステムと組み合わせて使い、プロデューサーとコンシューマー間のスキーマ互換性を保証します。

• スキーマのバージョン管理: 新フィールド追加やフィールド削除がいつ行われたか追跡可能
• 互換性チェック: BACKWARD / FORWARD / FULL の3モードで互換性を自動検証
• シリアライズ/デシリアライズ: スキーマIDをメッセージに埋め込み、コンシューマー側でスキーマを取得してデシリアライズ

Confluent Schema Registryとの連携

DatabricksからConfluentのKafkaトピックをAvroで読む場合、from_avroとschema registry URLを組み合わせてデシリアライズします。

from pyspark.sql.functions import col
from pyspark.sql.avro.functions import from_avro

schema_registry_options = {
    "schema.registry.url": "https://your-registry.confluent.cloud",
    "confluent.schema.registry.basic.auth.credentials.source": "USER_INFO",
    "confluent.schema.registry.basic.auth.user.info": "<API_KEY>:<API_SECRET>"
}

df = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "orders-topic")
    .option("startingOffsets", "latest")
    .load()
    .select(
        from_avro(
            col("value"),
            "orders-value",
            schema_registry_options
        ).alias("data")
    )
    .select("data.*")
)

from_avroの第2引数はSchema Registry上のSubject名（通常は topic名-value）です。 スキーマIDがメッセージヘッダーに含まれるため、プロデューサー側でスキーマが更新されても、レジストリからスキーマを動的に取得してデシリアライズできます。

Auto Loaderのスキーマ推論と進化

Auto Loader（cloudFiles）はクラウドストレージ上のファイルを増分取り込みする仕組みで、スキーマの自動推論と進化に対応しています。 Kafkaではなくファイルベースの取り込みにおいて、スキーマ管理の主力となる機能です。

df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.inferColumnTypes", "true")
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    .option("cloudFiles.schemaLocation", "/mnt/schema/orders")
    .load("/mnt/landing/orders/")
)

schemaLocationに保存されたスキーマファイルは、次回ストリーム起動時に読み込まれます。 addNewColumnsモードでは新カラム検出時にストリームが一時的にrestartされ、スキーマファイルが更新されます。

Delta Lakeのスキーマ進化

Deltaテーブルへの書き込み時にスキーマの差分をどう処理するかは、mergeSchemaとoverwriteSchemaで制御します。

# mergeSchema: 新カラムを追加（既存カラムは保持）
(df.writeStream
    .format("delta")
    .option("mergeSchema", "true")
    .option("checkpointLocation", "/mnt/checkpoints/orders")
    .outputMode("append")
    .toTable("silver.orders")
)

# overwriteSchema: スキーマ全体を置換（破壊的変更）
(df.write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("silver.orders_v2")
)

• mergeSchema: カラム追加のみ。型変更やカラム削除は不可。ストリーミングでもバッチでも使用可能
• overwriteSchema: スキーマ全体を置換。バッチのoverwriteモード限定。既存データの型と整合しなくなるリスクがあるため、テーブル再構築時のみ使用

Avro vs JSON: フォーマット選択基準

エンドツーエンドのスキーマ管理パターン

Kafkaソース＋Auto Loader＋Delta Lakeを組み合わせた典型的なスキーマ管理フローを示します。

[Producer]
  │  Avro + Schema Registry (BACKWARD互換)
  v
[Kafka Topic]
  │  from_avro + schema_registry_options
  v
[Databricks Streaming Job]
  │  スキーマ変更検出
  ├─ 新カラム → mergeSchema=true でDeltaに追加
  └─ 型変更  → ストリーム停止 → 手動スキーマ移行
       v
[Delta Table (Silver)]
  │  NOT NULL / CHECK制約で最終検証
  v
[Delta Table (Gold)]

[Cloud Storage (JSON/CSV/Parquet)]
  │  Auto Loader (cloudFiles)
  │  schemaEvolutionMode = addNewColumns
  │  schemaLocation = /mnt/schema/xxx
  v
[Databricks Streaming Job]
  │  mergeSchema=true
  v
[Delta Table (Bronze → Silver)]

運用上の注意点

• schemaLocationとcheckpointLocationは別パスを推奨。チェックポイントをリセットしてもスキーマは保持される
• addNewColumnsモードでは下流のテーブル・ビューにもカラム追加が波及するため、影響範囲を事前に把握する
• Schema Registryの互換性モードをBACKWARDにしておけば、コンシューマー（Databricks）側は新スキーマで古いメッセージも読める
• 型変更（int→string等）はmergeSchemaでは吸収できないため、overwriteSchemaまたはテーブル再構築が必要
• _rescued_data カラムのサイズを定期監視し、想定外のスキーマ変更を早期検出する

問題で確認

よくある質問