大量テーブルのデータパイプラインをコード量少なくスケールさせる設計
この文章はAIを使用して作成しています。
概要
データエンジニアリングの現場では、取り込むテーブルが数百・数千に増えるにつれてパイプラインの管理が破綻しがちです。本記事では、設定駆動（Config-driven）パイプライン という設計アプローチを使って、Databricks上で大量テーブルをコード量少なくスケールさせる方法を解説します。
よくある問題：テーブルが増えるたびにコードが増える
テーブルごとにパイプラインを書いていると、次のような問題が積み重なっていきます。
似たコードがあちこちに生まれる
# orders用パイプライン
def ingest_orders():
df = spark.read.format("jdbc").option("url", ...).option("query", "SELECT * FROM orders").load()
df.write.format("delta").mode("append").saveAsTable("bronze.orders")
# contacts用パイプライン（ほぼ同じ）
def ingest_contacts():
df = spark.read.format("jdbc").option("url", ...).option("query", "SELECT * FROM contacts").load()
df.write.format("delta").mode("append").saveAsTable("bronze.contacts")
# events用パイプライン（またほぼ同じ）
def ingest_events():
...
テーブルが10個になれば10個のほぼ同じ関数ができあがります。接続先URLが変わったときは10箇所を修正しなければならず、修正漏れがバグの原因になります。
新しいテーブルを追加するたびにPythonファイルを書く
テーブルを1つ追加するだけなのに、Pythonファイルを新規作成してレビューを通してデプロイする、というフローが毎回発生します。作業コストが高く、ミスも起きやすくなります。
品質チェック・エラーハンドリングが各パイプラインにバラバラに実装される
品質チェックのロジックを直したいとき、全パイプラインを修正して回る必要があります。チームが大きくなるほど、実装のばらつきが顕在化します。
前提①：ブロンズ・シルバー・ゴールドレイヤーとは
解決策の説明に入る前に、本記事で使うレイヤーの概念を整理しておきます。
データレイクアーキテクチャでは、データの加工度合いに応じて3つの層（レイヤー）に分けて管理します。
ソース（DB・API・ファイル）
│
▼
ブロンズ層：生データをそのまま保存する。変換しない
│
▼
シルバー層：クレンジング・結合済みのデータ。分析に使える状態
│
▼
ゴールド層：ビジネスロジックで集計済みのデータ。レポートやBIに直結
層を分けることで「どこまで加工されたデータか」が明確になり、問題が起きたときに原因を切り分けやすくなります。
レイヤーによって変換パターンが異なります。
| レイヤー | 変換パターン | 理由 |
|---|---|---|
| ブロンズ | 1対1のみ | ソースを忠実に取り込む。変換しない |
| シルバー | ファネル（多対1） | ブロンズテーブルを結合・クレンジングする |
| ゴールド | ファネル（多対1） | シルバーテーブルを集計・ビジネスロジックで変換する |
ここでいうファネル とは、複数のテーブルを1つのテーブルに集約する変換パターンです。
bronze.orders ─→ silver.orders ─╮
├─→ gold.order_summary（ファネル）
bronze.contacts ─→ silver.contacts ─╯
なお、ファンアウト（1つのソースから複数のテーブルを作るパターン）は、同じソースを参照するファネルを独立して複数定義することで表現できます。専用のパターンとして持つ必要はありません。
前提②：## 技術スタック
本記事では以下の技術スタックを前提とします。
| 技術 | 用途 |
|---|---|
| Databricks | 実行環境。SparkクラスタとDelta Lakeを提供する |
| Databricks Jobs | パイプラインのスケジュール実行・並列実行・依存関係管理を担う |
| Delta Lake | 書き込み先のテーブル形式。更新・削除・Upsertに対応している |
| PySpark | データ処理。SparkのDataFrameは遅延評価のため、大量データも分散処理できる |
| Python | パイプラインのロジック実装言語 |
| YAML | パイプラインの設定ファイル形式 |
| GitHub Actions / Databricks Asset Bundle | CI/CDからJobを登録・更新する |
Databricks Jobsを使うことで、テーブル間の依存関係を宣言的に定義しながら並列実行できます。1つのタスクが失敗しても他のタスクには影響せず、リトライも個別にかけられます。
解決策：設定駆動パイプライン
設定駆動の考え方はシンプルです。「テーブルごとに異なるのは設定だけ。処理のコードは1つに共通化する」 。
たとえばPostgreSQLの orders テーブルを取り込みたい場合、書くのはこれだけです。
tables:
- name: raw_orders
source_type: postgres
query: "SELECT * FROM orders WHERE updated_at > '{last_run}'"
destination: bronze.orders
schedule: "0 * * * *" # 毎時0分
load_mode: append
quality:
on_failure: error
checks:
- type: not_null
columns: [order_id]
新しいテーブルを追加するときはこのエントリをYAMLに1つ追加するだけです。Pythonコードには触れません。source_type を変えればSalesforceやS3にも対応でき、load_mode で全量洗い替え・差分更新・追記を切り替えられます。schedule でテーブルごとに実行タイミングを個別に設定でき、同じスケジュールのテーブルは自動的に同一Jobにまとめられます。品質チェックの設定も同じファイルに書くだけで自動的に実行されます。
品質チェックのロジックを直したいときは共通モジュールの1ファイルだけ修正すれば全テーブルに反映されます。詳細は後続のセクションで説明します。
全体構造
各層の処理の流れを整理します。ブロンズ層は生データをそのまま保存するだけなので変換はありません。シルバー/ゴールド層はブロンズやシルバーのテーブルを複数参照して結合・変換する処理が加わります。
ブロンズ層
ソース（PostgreSQL・Salesforce・S3など）
│ Extract ← ソースからデータを取得
▼
生データ
│ Quality Check ← データ品質チェック
▼
品質確認済みデータ
│ Load ← Deltaテーブルに書き込む
▼
bronze.テーブル（Delta Table）
シルバー/ゴールド層 ブロンズ層の処理にTransformが加わります。
bronze/silver.テーブルA bronze/silver.テーブルB …
│ │
└───────────┬────────────┘
│ Extract ← 複数ソースを取得
▼
複数のDataFrame
│ Transform ← 結合・クレンジング・集計
▼
変換済みDataFrame
│ Quality Check
▼
品質確認済みデータ
│ Load
▼
silver/gold.テーブル（Delta Table）
Extract・Quality Check・Loadの実装は共通モジュール（sources/ / quality.py / loader.py）として両層で共有します。「どのソースから取得するか」「どんなチェックを行うか」「どのパターンで書き込むか」はすべてYAMLの設定に書かれており、ノートブックが設定を読んで処理を切り替えます。コードを変えることなく、設定だけで動作を制御できます。
役割ごとにディレクトリが独立しているため、どこを修正すればよいかが一目でわかります。
pipeline/
├── config/
│ ├── models.py # YAMLを読み込むデータオブジェクトの定義
│ ├── bronze_config.yaml # ブロンズ層：取り込むテーブルを宣言
│ └── silver_gold_config.yaml # シルバー/ゴールド層：変換するテーブルを宣言
├── sources/
│ ├── base.py # 「データを取得する」インターフェース定義
│ ├── postgres.py # PostgreSQLからの取得実装
│ ├── salesforce.py # Salesforceからの取得実装
│ ├── s3.py # S3からの取得実装
│ ├── delta.py # DeltaTableからの取得実装
│ └── registry.py # どのソースタイプにどのクラスを使うかの対応表
├── transforms/
│ ├── orders.py # ordersテーブルの変換ロジック
│ └── order_summary.py # order_summaryテーブルの変換ロジック
├── quality.py # データ品質チェック（共通）
├── loader.py # Deltaテーブルへの書き込み（共通）
├── job_deployer.py # Databricks JobsへのJob登録ロジック（共通）
└── deploy.py # CI/CDから実行するエントリーポイント
共通モジュール
ブロンズ・シルバー/ゴールドの両層で共通して使うモジュールです。ここを理解しておくと、後続の各層の実装がスムーズに読めます。
config/models.py — 設定のデータオブジェクト
YAMLを読み込んだ結果を dict のままコードに流すと、タイポや設定漏れが実行時まで気づけず、各関数にどのキーが必要かも把握しづらくなります。dataclass でデータオブジェクトとして定義することで、IDEの補完が効き、コードの意図が明確になります。
from dataclasses import dataclass, field
@dataclass
class QualityCheck:
type: str
columns: list[str] = field(default_factory=list)
threshold: int = 0
column: str = ""
min: float | None = None
@dataclass
class QualityConfig:
on_failure: str = "error"
checks: list[QualityCheck] = field(default_factory=list)
@dataclass
class SourceConfig:
type: str
alias: str
url: str = ""
table: str = ""
query: str = ""
path: str = ""
@dataclass
class TableConfig:
name: str
destination: str
load_mode: str = "overwrite"
quality: QualityConfig = field(default_factory=QualityConfig)
# ブロンズ層用
source_type: str = ""
schedule: str = "" # cron形式。例: "0 * * * *"（毎時）、"0 6 * * *"（日次6時）
# シルバー/ゴールド層用
sources: list[SourceConfig] = field(default_factory=list)
transform: str = ""
depends_on: list[str] = field(default_factory=list)
merge_keys: list[str] = field(default_factory=list)
def load_config(path: str) -> list[TableConfig]:
import yaml
from dacite import from_dict
with open(path) as f:
raw = yaml.safe_load(f)
return [from_dict(TableConfig, t) for t in raw["tables"]]
dacite はネストされた辞書を dataclass に変換するライブラリです。バリデーションを強化したい場合は発展セクションで pydantic への移行方法を紹介しています。
sources/ — データを取得する
データソースごとの取得ロジックをクラスとして切り出します。ExtractorProtocol はすべてのExtractorが実装すべきメソッドを定義するインターフェースです。Pythonの Protocol を使うことで継承が不要になり、既存のクラスもそのまま組み込めます。
sources/base.py
from typing import Protocol
from pyspark.sql import DataFrame
from config.models import SourceConfig
class ExtractorProtocol(Protocol):
def extract(self, config: SourceConfig) -> DataFrame:
...
sources/postgres.py
from pyspark.sql import DataFrame
from config.models import SourceConfig
class PostgresExtractor:
def extract(self, config: SourceConfig) -> DataFrame:
return (
spark.read.format("jdbc")
.option("url", config.url)
.option("query", config.query)
.load()
)
「どのソースタイプにどのクラスを使うか」という対応表は registry.py に集約します。新しいデータソースを追加するときは、Extractorクラスを sources/ に追加して registry.py に1行書き加えるだけです。ノートブックや他のモジュールを修正する必要はありません。
たとえばBigQueryを追加したい場合はこうなります。
sources/registry.py
from sources.base import ExtractorProtocol
from sources.postgres import PostgresExtractor
from sources.salesforce import SalesforceExtractor
from sources.s3 import S3Extractor
from sources.delta import DeltaExtractor
# from sources.bigquery import BigQueryExtractor ← 追加するだけ
EXTRACTORS: dict[str, ExtractorProtocol] = {
"postgres": PostgresExtractor(),
"salesforce": SalesforceExtractor(),
"s3": S3Extractor(),
"delta": DeltaExtractor(),
# "bigquery": BigQueryExtractor(), ← 追加するだけ
}
このような書き方はストラテジーパターンと呼ばれます。if/elif の連鎖で分岐を書く代わりに、処理をクラスに閉じ込めて辞書で切り替える方法で、拡張しやすく読みやすいコードになります。新しいパターンを追加するときはクラスを1つ書いて EXTRACTORS に1行登録するだけで、既存コードには触れません。
シルバー/ゴールド層ではブロンズのDelta Tableを読み込むことが多いため、DeltaExtractor の実装例も示しておきます。
sources/delta.py
from pyspark.sql import DataFrame
from config.models import SourceConfig
class DeltaExtractor:
def extract(self, config: SourceConfig) -> DataFrame:
return spark.read.format("delta").table(config.table)
quality.py — データ品質チェック
設定ファイルに宣言されたチェックを実行します。on_failure: error はチェック失敗時にパイプラインを止め、on_failure: warn は警告ログを出して続行します。
チェックの種類は辞書（CHECKERS）で管理します。sources/、registry.pyと同じストラテジーパターンで実装しており、新しいチェックを追加するときはクラスを1つ書いて CHECKERS に1行登録するだけで、既存コードには触れません。
import logging
from typing import Protocol, TypedDict
from pyspark.sql import DataFrame
from config.models import TableConfig, QualityCheck
logger = logging.getLogger(__name__)
class CheckResult(TypedDict):
passed: bool
message: str
class CheckerProtocol(Protocol):
def check(self, df: DataFrame, rule: QualityCheck) -> CheckResult:
...
class NotNullChecker:
def check(self, df: DataFrame, rule: QualityCheck) -> CheckResult:
for col in rule.columns:
null_count: int = df.filter(df[col].isNull()).count()
if null_count > 0:
return {"passed": False, "message": f"{col} に {null_count} 件のNULLがあります"}
return {"passed": True, "message": ""}
class MinRowsChecker:
def check(self, df: DataFrame, rule: QualityCheck) -> CheckResult:
row_count: int = df.count()
if row_count < rule.threshold:
return {"passed": False, "message": f"行数が不足しています（{row_count} < {rule.threshold}）"}
return {"passed": True, "message": ""}
class ValueRangeChecker:
def check(self, df: DataFrame, rule: QualityCheck) -> CheckResult:
if rule.min is not None:
violation: int = df.filter(df[rule.column] < rule.min).count()
if violation > 0:
return {"passed": False, "message": f"{rule.column} に範囲外の値が {violation} 件あります"}
return {"passed": True, "message": ""}
CHECKERS: dict[str, CheckerProtocol] = {
"not_null": NotNullChecker(),
"min_rows": MinRowsChecker(),
"value_range": ValueRangeChecker(),
}
def run_checks(df: DataFrame, config: TableConfig) -> None:
on_failure: str = config.quality.on_failure
for rule in config.quality.checks:
checker = CHECKERS.get(rule.type)
if checker is None:
raise ValueError(f"未対応のチェック: {rule.type}")
result: CheckResult = checker.check(df, rule)
if not result["passed"]:
message: str = f"[{config.name}] 品質チェック失敗: {result['message']}"
if on_failure == "error":
raise ValueError(message)
else:
logger.warning(message)
loader.py — Deltaテーブルへの書き込み
書き込み処理を一元管理するモジュールです。全量洗い替え・差分更新・追記・パーティション単位の洗い替えといった複数の書き込みパターンをストラテジーパターンで実装します。
from typing import Protocol
from pyspark.sql import DataFrame
from delta.tables import DeltaTable
from config.models import TableConfig
class LoaderProtocol(Protocol):
def load(self, df: DataFrame, config: TableConfig) -> None:
...
class OverwriteLoader:
def load(self, df: DataFrame, config: TableConfig) -> None:
df.write.format("delta").mode("overwrite").saveAsTable(config.destination)
class AppendLoader:
def load(self, df: DataFrame, config: TableConfig) -> None:
df.write.format("delta").mode("append").saveAsTable(config.destination)
class MergeLoader:
def load(self, df: DataFrame, config: TableConfig) -> None:
merge_condition = " AND ".join([f"target.{k} = source.{k}" for k in config.merge_keys])
delta_table = DeltaTable.forName(spark, config.destination)
(
delta_table.alias("target")
.merge(df.alias("source"), merge_condition)
.whenMatchedUpdateAll()
.whenNotMatchedInsertAll()
.execute()
)
class OverwritePartitionLoader:
def load(self, df: DataFrame, config: TableConfig) -> None:
(
df.write.format("delta")
.mode("overwrite")
.option("replaceWhere", f"{config.partition_column} = '{config.partition_value}'")
.saveAsTable(config.destination)
)
LOADERS: dict[str, LoaderProtocol] = {
"overwrite": OverwriteLoader(),
"append": AppendLoader(),
"merge": MergeLoader(),
"overwrite_partition": OverwritePartitionLoader(),
}
def load(df: DataFrame, config: TableConfig) -> None:
loader = LOADERS.get(config.load_mode)
if loader is None:
raise ValueError(f"未対応のload_mode: {config.load_mode}")
loader.load(df, config)
load_mode には4つのパターンを指定できます。必要に応じて、ロード処理のクラスを追加します。
| load_mode | 挙動 | ユースケース |
|---|---|---|
overwrite | 全量洗い替え | マスタデータ、小テーブル |
merge | キー一致で更新、なければ追加（Upsert） | トランザクションデータ、変更が発生するデータ |
append | 既存データに追記するだけ | ログ、イベントデータ |
overwrite_partition | 指定パーティションだけ洗い替え | 日次バッチで特定日付分だけ更新したい場合 |
ブロンズ層：1対1パイプライン
ソースを忠実に取り込むだけで、変換は行いません。source_type でどのExtractorを使うか指定し、そのままDeltaテーブルに書き込みます。
config/bronze_config.yaml
tables:
- name: raw_orders
source_type: postgres
query: "SELECT * FROM orders WHERE updated_at > '{last_run}'"
destination: bronze.orders
schedule: "0 * * * *" # 毎時0分
load_mode: append # 新しいレコードを追記していく
quality:
on_failure: error # チェック失敗時はパイプラインを止める
checks:
- type: not_null
columns: [order_id]
- type: min_rows
threshold: 1
- name: raw_contacts
source_type: salesforce
source_object: Contact
destination: bronze.contacts
schedule: "0 * * * *" # 毎時0分（raw_ordersと同じJobにまとめられる）
load_mode: overwrite # 毎回全量洗い替え
quality:
on_failure: warn # チェック失敗時は警告ログのみで続行
checks:
- type: not_null
columns: [Id]
- name: raw_events
source_type: s3
path: s3://my-bucket/events/
destination: bronze.events
schedule: "0 6 * * *" # 毎日6時（別Jobになる）
load_mode: append
quality:
on_failure: warn
checks:
- type: not_null
columns: [event_id]
bronze_task ノートブック
deploy.py で登録された各Jobタスクはノートブックを呼び出します。テーブルごとに独立したタスクとして並列実行されるため、1つが失敗しても他に影響しません。table_name をパラメータとして受け取り、設定を読み込んでExtract → Quality Check → Loadを実行します。
from pyspark.sql import DataFrame
from sources.registry import EXTRACTORS
from quality import run_checks
from loader import load
from config.models import TableConfig, load_config
# Databricks Jobsからパラメータとして渡されるテーブル名を受け取る
table_name: str = dbutils.widgets.get("table_name")
table: TableConfig = next(t for t in load_config("config/bronze_config.yaml") if t.name == table_name)
# Extract：source_typeに対応するExtractorを取り出して実行
df: DataFrame = EXTRACTORS[table.source_type].extract(table)
# Quality Check
run_checks(df, table)
# Load
load(df, table)
シルバー/ゴールド層：ファネルパイプライン
複数のテーブルを参照して結合・変換します。すべての変換パターンをファネル（多対1）として表現します。ソースが1つの場合もファネルとして書くことで、後からソースを追加してもtransformの関数定義を変えなくて済みます。
config/silver_gold_config.yaml
depends_on に依存するテーブル名を列挙します。依存のないテーブルはDatabricks Jobsによって並列実行され、依存があるテーブルは依存先が完了してから実行されます。
tables:
- name: silver_orders
depends_on: [] # 依存なし → ブロンズ完了後すぐ実行
sources:
- type: delta
table: bronze.orders
alias: orders # transformの中でdfs["orders"]として参照
destination: silver.orders
transform: transforms.orders.clean
load_mode: merge
merge_keys: [order_id]
quality:
on_failure: error
checks:
- type: not_null
columns: [order_id, amount]
- type: min_rows
threshold: 1000
- type: value_range
column: amount
min: 0
- name: silver_contacts
depends_on: [] # 依存なし → silver_ordersと同時に実行される
sources:
- type: delta
table: bronze.contacts
alias: contacts
destination: silver.contacts
transform: transforms.contacts.clean
load_mode: overwrite
quality:
on_failure: warn
checks:
- type: not_null
columns: [contact_id]
- name: gold_order_summary
depends_on: [silver_orders, silver_contacts] # 両方完了後に実行
sources:
- type: delta
table: silver.orders
alias: orders
- type: delta
table: silver.contacts
alias: contacts
destination: gold.order_summary
transform: transforms.order_summary.build
load_mode: overwrite
quality:
on_failure: error
checks:
- type: not_null
columns: [order_id, contact_id]
transforms/ — 変換ロジック
transformの関数はすべて dfs: dict[str, DataFrame] を受け取り、1つの DataFrame を返す形に統一します。dfs の各キーは設定ファイルの alias に対応します。ソースが1つの場合も同じシグネチャで書くことで、後からソースを追加してもtransformの関数定義は変えなくて済みます。
transforms/orders.py — ソース1つのファネル
from pyspark.sql import DataFrame
from pyspark.sql import functions as F
from pyspark.sql.window import Window
def clean(dfs: dict[str, DataFrame]) -> DataFrame:
df = dfs["orders"] # aliasで指定したキーで取り出す
return (
df
.drop("_internal_id", "_raw_payload")
.withColumn("amount", F.col("amount").cast("double"))
.withColumn("ordered_at", F.to_timestamp("ordered_at", "yyyy-MM-dd HH:mm:ss"))
.filter(F.col("order_id").isNotNull())
.filter(F.col("amount") > 0)
.withColumn("status", F.lower(F.trim(F.col("status"))))
.withColumn("order_year", F.year("ordered_at"))
.withColumn("order_month", F.month("ordered_at"))
.withColumn(
"row_num",
F.row_number().over(
Window.partitionBy("order_id").orderBy(F.desc("ordered_at"))
)
)
.filter(F.col("row_num") == 1)
.drop("row_num")
)
transforms/order_summary.py — 複数ソースのファネル
from pyspark.sql import DataFrame
def build(dfs: dict[str, DataFrame]) -> DataFrame:
orders = dfs["orders"]
contacts = dfs["contacts"]
return (
orders
.join(contacts, orders["customer_id"] == contacts["Id"], "left")
.select("order_id", "contact_id", "amount", "ordered_at")
)
silver_gold_task ノートブック
table_name をパラメータとして受け取り、設定を読み込んでExtract → Transform → Quality Check → Loadを実行します。silver_orders と silver_contacts は並列実行され、両方完了後に gold_order_summary が実行されます。
silver_orders ─╮
├─→ gold_order_summary
silver_contacts ─╯
import importlib
from types import ModuleType
from typing import Callable
from pyspark.sql import DataFrame
from sources.registry import EXTRACTORS
from quality import run_checks
from loader import load
from config.models import TableConfig, load_config
# Databricks Jobsからパラメータとして渡されるテーブル名を受け取る
table_name: str = dbutils.widgets.get("table_name")
table: TableConfig = next(t for t in load_config("config/silver_gold_config.yaml") if t.name == table_name)
# Extract：複数ソースをaliasをキーにした辞書として取得
# SparkのDataFrameは遅延評価のため、この時点ではデータを読み込まない
# 実際にデータが処理されるのはtransform内でアクションが実行されたとき
dfs: dict[str, DataFrame] = {source.alias: EXTRACTORS[source.type].extract(source.table) for source in table.sources}
# Transform：YAMLの "transforms.orders.clean" という文字列からモジュールと関数名を分解して動的にロード
# → transforms/orders.py の clean() 関数を実行時に呼び出す
module_path, func_name = table.transform.rsplit(".", 1)
module: ModuleType = importlib.import_module(module_path)
transform_fn: Callable[[dict[str, DataFrame]], DataFrame] = getattr(module, func_name)
df: DataFrame = transform_fn(dfs)
# Quality Check
run_checks(df, table)
# Load
load(df, table)
デプロイ：Databricks Jobsへの登録
Jobの登録はCI/CDパイプライン（GitHub ActionsやDatabricks Asset Bundleなど）から行います。設定ファイルが変わったとき（テーブルの追加・削除など）に再実行してJobを更新します。
登録ロジックは job_deployer.py にまとめます。bronze_config.yaml の schedule をキーにブロンズをグループ化し、スケジュールごとに1つのJobを作成します。各Jobには、対応するブロンズに依存するシルバー/ゴールドのタスクが後続として自動的に追加されます。全テーブルが同じノートブックを共有し、table_name パラメータで処理するテーブルを切り替えるため、テーブルが100個あってもノートブックは2つだけで済みます。
登録後のJobのタスクグラフは次のようになります（raw_orders・raw_contacts が毎時、raw_events が日次の場合）。
pipeline_0_*_*_*_*（毎時）
├── [Bronze] raw_orders
├── [Bronze] raw_contacts
├── [Silver] silver_orders depends_on: [raw_orders]
└── [Silver] silver_contacts depends_on: [raw_contacts]
pipeline_0_6_*_*_*（日次）
├── [Bronze] raw_events
├── [Silver] silver_events depends_on: [raw_events]
└── [Gold] gold_order_summary depends_on: [silver_orders, silver_contacts, silver_events, raw_events]
↑ raw_events を depends_on に書くことでこのJobに割り当てられる
↑ silver_orders・silver_contacts は毎時Jobで完了済みの前提で参照
silver_gold_config.yaml の depends_on の書き方
depends_on にはシルバー/ゴールドのテーブル名（タスク順序）とブロンズのテーブル名（Jobグループの割り当て）の両方を書けます。ブロンズ名を書いたエントリは「そのブロンズが動くスケジュールのJobに自分を入れる」という意味を持ちます。
# 同じスケジュールのブロンズだけに依存する場合
- name: silver_orders
depends_on: [raw_orders] # ブロンズ名 → 毎時Jobに入る＋タスク順序
# 別スケジュールのブロンズを待つ場合
- name: gold_order_summary
depends_on: [silver_orders, silver_contacts, silver_events, raw_events]
# ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ S/G間の依存（タスク順序）
# ^^^^^^^^^^
# ブロンズ名 → 日次Jobに入る
job_deployer.py
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import (
TaskDependency,
NotebookTask,
Task,
CronSchedule,
)
from config.models import TableConfig
def _bronze_names(bronze_tables: list[TableConfig]) -> set[str]:
"""ブロンズのタスク名セットを返す。例: {"raw_orders", "raw_contacts"}"""
return {t.name for t in bronze_tables}
def _sg_tables_for_group(
sg_tables: list[TableConfig],
group_bronze_names: set[str],
all_bronze_names: set[str],
) -> list[TableConfig]:
"""このブロンズグループに含めるべきシルバー/ゴールドを返す。
depends_on にこのグループのブロンズ名が含まれるものを起点に、
S/G 間の depends_on を幅優先探索して推移的な依存も追跡する。
"""
# depends_on にこのグループのブロンズ名が直接含まれるものを起点にする
directly_dependent: set[str] = {
t.name
for t in sg_tables
if group_bronze_names & set(t.depends_on)
}
# 推移的に依存するもの（BFS）
# S/G 間の depends_on のみを辿る（ブロンズ名はグループ判定に使うため除外）
included: set[str] = set(directly_dependent)
frontier = list(directly_dependent)
while frontier:
current = frontier.pop()
for t in sg_tables:
if t.name in included:
continue
sg_deps = [d for d in t.depends_on if d not in all_bronze_names]
if current in sg_deps:
included.add(t.name)
frontier.append(t.name)
return [t for t in sg_tables if t.name in included]
def _build_bronze_tasks(
bronze_tables: list[TableConfig],
notebook_path: str,
) -> list[Task]:
return [
Task(
task_key=t.name,
depends_on=[],
notebook_task=NotebookTask(
notebook_path=notebook_path,
base_parameters={"table_name": t.name},
),
)
for t in bronze_tables
]
def _build_sg_tasks(
sg_tables: list[TableConfig],
notebook_path: str,
) -> list[Task]:
"""シルバー/ゴールドのタスクを構築する。
depends_on は YAML に書いた値をそのまま TaskDependency に変換する。
ブロンズ名も S/G 名もどちらもタスク依存として扱われ、
Databricks Jobs がタスクグラフ通りに順序を制御する。
"""
return [
Task(
task_key=t.name,
depends_on=[TaskDependency(task_key=dep) for dep in t.depends_on],
notebook_task=NotebookTask(
notebook_path=notebook_path,
base_parameters={"table_name": t.name},
),
)
for t in sg_tables
]
def deploy_jobs_by_schedule(
bronze_tables: list[TableConfig],
sg_tables: list[TableConfig],
bronze_notebook_path: str,
sg_notebook_path: str,
) -> None:
"""スケジュール単位でブロンズをグループ化し、
対応するシルバー/ゴールドを後続タスクとして同一 Job に登録する。
"""
all_bronze_names = _bronze_names(bronze_tables)
# schedule をキーにグループ化
schedule_groups: dict[str, list[TableConfig]] = {}
for t in bronze_tables:
schedule_groups.setdefault(t.schedule, []).append(t)
client = WorkspaceClient()
for schedule, group_bronze in schedule_groups.items():
group_bronze_names = _bronze_names(group_bronze)
group_sg = _sg_tables_for_group(sg_tables, group_bronze_names, all_bronze_names)
tasks: list[Task] = (
_build_bronze_tasks(group_bronze, bronze_notebook_path)
+ _build_sg_tasks(group_sg, sg_notebook_path)
)
job_name = f"pipeline_{schedule.replace(' ', '_')}"
client.jobs.create(
name=job_name,
tasks=tasks,
schedule=CronSchedule(
quartz_cron_expression=schedule,
timezone_id="Asia/Tokyo",
),
)
deploy.py （CI/CDから実行）
from job_deployer import deploy_jobs_by_schedule
from config.models import load_config
deploy_jobs_by_schedule(
bronze_tables=load_config("config/bronze_config.yaml"),
sg_tables=load_config("config/silver_gold_config.yaml"),
bronze_notebook_path="/notebooks/bronze_task",
sg_notebook_path="/notebooks/silver_gold_task",
)
まとめ
設定駆動パイプラインを導入することで、冒頭で挙げた問題がこう解決されます。
似たコードがあちこちに生まれる問題 Extract・Quality・Loadの処理は共通モジュールに1度だけ実装します。接続先URLが変わっても修正箇所は1ファイルです。
新しいテーブルを追加するたびにPythonファイルを書く問題 YAMLに1エントリ追加するだけで済みます。レビューもPythonコードではなく設定ファイルを見るだけです。
品質チェック・エラーハンドリングが各パイプラインにバラバラに実装される問題 quality.py と loader.py を共通モジュールとして集約しています。ストラテジーパターンで実装しているため、新しいチェックや書き込みパターンを追加しても既存コードには触れません。
変更対応の一覧をまとめます。
| やりたいこと | 対応箇所 |
|---|---|
| ブロンズテーブルを追加する | bronze_config.yaml に1エントリ追加するだけ |
| シルバー/ゴールドテーブルを追加する | silver_gold_config.yaml に1エントリ追加するだけ |
| 複数ソースを結合する | sources リストにエントリを追加し、transformで dfs["alias"] で取り出す |
| ソースの種類を増やす | sources/ にExtractorを追加し registry.py に登録するだけ |
| 変換ロジックを変える | transforms/ の対象ファイルだけ修正 |
| loadパターンを変える | 各configの load_mode を切り替えるだけ |
| 品質チェックを追加・変更する | 各configの checks を編集するだけ |
| チェック失敗時の挙動を変える | 各configの on_failure を error / warn で切り替えるだけ |
| テーブル間の依存関係を定義する | silver_gold_config.yaml の depends_on を編集するだけ |
| ブロンズのスケジュールを変える | bronze_config.yaml の schedule を変えるだけ。同じ値のブロンズは自動的に同一Jobにまとめられる |
| 別スケジュールのブロンズを待ってからS/Gを動かす | silver_gold_config.yaml の depends_on に待ちたいブロンズのテーブル名を追加するだけ |
| 並列化する | deploy.py をCI/CDから実行してDatabricks Jobsにタスクグラフを登録する |
発展：さらにロバストにするために
本筋の実装が固まったら、次の2点を追加することでパイプラインの信頼性を高められます。
pydantic によるバリデーション
本記事では dataclass で設定オブジェクトを定義しましたが、pydantic に移行するとYAML読み込み時点でバリデーションが走り、設定ミスをパイプラインが途中で落ちる前に検知できます。
from pydantic import BaseModel, field_validator
from typing import Literal
class QualityConfig(BaseModel):
on_failure: Literal["error", "warn"] = "error" # ← error/warn以外はバリデーションエラー
checks: list[QualityCheck] = []
class TableConfig(BaseModel):
name: str
destination: str
load_mode: Literal["overwrite", "append", "merge", "overwrite_partition"] = "overwrite" # ← タイポはここで検知
...
@field_validator("merge_keys")
@classmethod
def merge_keys_required_for_merge(cls, v, info):
if info.data.get("load_mode") == "merge" and not v:
raise ValueError("load_mode=merge には merge_keys の指定が必要です")
return v
def load_config(path: str) -> list[TableConfig]:
import yaml
with open(path) as f:
raw = yaml.safe_load(f)
return [TableConfig(**t) for t in raw["tables"]] # ← ここでバリデーションが走る
Literal で許容する値を列挙することで、load_mode: mereg のようなタイポや on_failure: stop のような不正な値をYAML読み込み時点でエラーとして検知できます。dacite が不要になり、pydantic 単体で変換とバリデーションを担います。
チェックの追加例
ストラテジーパターンの拡張性を活かして、独自のチェックを追加できます。たとえば「重複行がないか」をチェックする NoDuplicatesChecker を追加したい場合はこうなります。
# 1. クラスを追加する
class NoDuplicatesChecker:
def check(self, df: DataFrame, rule: QualityCheck) -> CheckResult:
total: int = df.count()
distinct: int = df.select(rule.column).distinct().count()
if total != distinct:
return {"passed": False, "message": f"{rule.column} に {total - distinct} 件の重複があります"}
return {"passed": True, "message": ""}
# 2. CHECKERSに1行追加する
CHECKERS: dict[str, CheckerProtocol] = {
"not_null": NotNullChecker(),
"min_rows": MinRowsChecker(),
"value_range": ValueRangeChecker(),
"no_duplicates": NoDuplicatesChecker(), # ← 追加するだけ
}
追加後はYAMLの checks に type: no_duplicates と書くだけで使えます。