自前のサーバーにブログを移行しました。
大量テーブルのデータパイプラインをコード量少なくスケールさせる設計
この文章は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 と書くだけで使えます。
【アーキテクチャパターン】 Pythonでリポジトリパターンを書いてみる
この文章はAIを使用して作成しています。
1. なぜリポジトリパターンが必要なのか
従来の課題：ビジネスロジックとデータアクセスの混在
リポジトリパターンを使わない場合、ビジネスロジックの中にデータアクセスのコードが直接書き込まれがちです。
# ❌ パターンを使わない場合 import sqlite3 def register_user(user_id: int, name: str, email: str) -> None: conn = sqlite3.connect("app.db") row = conn.execute("SELECT id FROM users WHERE id = ?", (user_id,)).fetchone() if row: raise ValueError(f"ユーザー {user_id} は既に存在します") conn.execute("INSERT INTO users VALUES (?, ?, ?)", (user_id, name, email)) conn.commit()
このコードには次の3つの問題があります。
1. 変更に弱い DBをSQLiteからPostgreSQLに変えたい場合、register_user を直接書き直さなければなりません。データアクセスの都合がビジネスロジックに波及してしまいます。
2. テストが難しい 関数の中でDBコネクションを直接生成しているため、テスト時にも実際のDBが必要になります。テストのたびにDBのセットアップ・クリーンアップが必要になり、テストが遅く・不安定になります。
3. 同じSQLが散らばる アプリが大きくなるにつれ、似たようなSQLがサービス層・コントローラー層などあちこちに重複して書かれていきます。修正漏れやバグの温床になります。
解決策：リポジトリパターン
リポジトリパターンは「データの保存・取得をどう行うか」をビジネスロジックから切り離し、専用のクラス（リポジトリ）に閉じ込める設計パターンです。
ポイントは2段階の分離です。
- 抽象リポジトリ（インターフェース）をドメイン層に置き、「何ができるか」だけを定義する
- 具体的な実装（SQLite、インメモリ等）をインフラ層に隔離し、「どうやるか」を閉じ込める
# ✅ リポジトリパターンを使う場合 class UserService: def __init__(self, repo: UserRepository): # 抽象に依存 self.repo = repo def register(self, user_id: int, name: str, email: str) -> None: if self.repo.find_by_id(user_id): raise ValueError(f"ユーザー {user_id} は既に存在します") self.repo.save(User(id=user_id, name=name, email=email))
UserService は UserRepository というインターフェースにしか依存していません。裏側がSQLiteなのかインメモリなのかを知らなくてよいため、次のことが可能になります。
- DBを別の種類に乗り換えても、
UserServiceには一切手を加えなくてよい - テスト時は
InMemoryUserRepositoryに差し替えるだけで、DBなしで高速にテストできる - データアクセスのコードがリポジトリクラスに集約されるため、SQLの重複が起きない
2. 全体構造を掴む
3つの層と役割
リポジトリパターンはアプリケーションを3つの層に分けて考えます。
| 層 | 役割 | 依存してよいもの |
|---|---|---|
| domain（ドメイン層） | ビジネスの概念を定義する。エンティティと抽象リポジトリを置く。 | なし（最も内側） |
| service（サービス層） | ビジネスロジックを実装する。抽象リポジトリを通じてデータを扱う。 | domain層のみ |
| infrastructure（インフラ層） | データの実際の保存・取得を担う。DBやAPIとのやりとりを閉じ込める。 | domain層のみ |
infrastructure は domain を知っていますが、domain は infrastructure を知りません。この一方向の依存がパターンの核心です。内側の層（domain）が外側の層（infrastructure）に依存してしまうと、DBを変えるたびにビジネスロジックの修正が必要になってしまいます。
┌─────────────────────────┐
│ domain 層 │
│ User（エンティティ） │
│ UserRepository（抽象） │
└────────────┬────────────┘
│ 依存
┌────────────────────┼───────────────────┐
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌────────────────┐ ┌───────────────────┐
│ service 層 │ │infrastructure層│ │ main.py │
│ UserService │ │ InMemory実装 │ │ どの実装を使うか │
│ │ │ SQLite実装 │ │ ここだけが知っている │
└───────────────┘ └────────────────┘ └───────────────────┘
ディレクトリ構成
層ごとにディレクトリを分けることで、依存関係がフォルダ構造として視覚化されます。
my_app/ ├── domain/ │ ├── user.py # エンティティ │ └── user_repository.py # 抽象リポジトリ ├── infrastructure/ │ ├── in_memory_user_repository.py # インメモリ実装（テスト・開発用） │ └── sqlite_user_repository.py # SQLite実装（本番用） ├── service/ │ └── user_service.py # ビジネスロジック └── main.py # エントリーポイント（実装の選択）
3. 実装を読む
domain層：ビジネスの概念を定義する
domain/user.py — エンティティ
アプリケーションの中心となるデータ構造です。データベースや通信の都合に依存せず、純粋にビジネス上の「ユーザー」という概念だけを表します。他のどのモジュールにも依存しない、最も内側の層です。
from dataclasses import dataclass @dataclass class User: id: int name: str email: str
domain/user_repository.py — 抽象リポジトリ
「ユーザーをどう保存・取得するか」のインターフェース（契約）だけを定義します。具体的な実装（SQLなのかAPIなのか）はここには一切登場しません。domain 層に置くことで、ビジネスロジックは「どんなデータソースでも同じように扱える」という前提で書けるようになります。
from abc import ABC, abstractmethod from domain.user import User class UserRepository(ABC): @abstractmethod def find_by_id(self, user_id: int) -> User | None: pass @abstractmethod def save(self, user: User) -> None: pass @abstractmethod def find_all(self) -> list[User]: pass
infrastructure層：実装の詳細を閉じ込める
同じ UserRepository インターフェースを、異なる技術で実装します。どちらを使っても service 層からは同じように扱えます。
infrastructure/in_memory_user_repository.py — インメモリ実装
辞書をストレージとして使うシンプルな実装です。DBのセットアップが不要なので、テストや開発初期に重宝します。UserRepository の契約さえ守っていれば、内部実装は何でも構いません。
from domain.user import User from domain.user_repository import UserRepository class InMemoryUserRepository(UserRepository): def __init__(self): self._store: dict[int, User] = {} def find_by_id(self, user_id: int) -> User | None: return self._store.get(user_id) def save(self, user: User) -> None: self._store[user.id] = user def find_all(self) -> list[User]: return list(self._store.values())
infrastructure/sqlite_user_repository.py — SQLite実装
同じインターフェースを今度はSQLiteで実装します。SQL文やコネクション管理といった「インフラの詳細」がこのクラスの中だけに閉じ込められているのがポイントです。将来PostgreSQLやMongoDBに乗り換えたとしても、このファイルを差し替えるだけで済みます。
import sqlite3 from domain.user import User from domain.user_repository import UserRepository class SQLiteUserRepository(UserRepository): def __init__(self, db_path: str): self.conn = sqlite3.connect(db_path) self.conn.execute( "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)" ) def find_by_id(self, user_id: int) -> User | None: row = self.conn.execute( "SELECT id, name, email FROM users WHERE id = ?", (user_id,) ).fetchone() return User(*row) if row else None def save(self, user: User) -> None: self.conn.execute( "INSERT OR REPLACE INTO users VALUES (?, ?, ?)", (user.id, user.name, user.email) ) self.conn.commit() def find_all(self) -> list[User]: rows = self.conn.execute("SELECT id, name, email FROM users").fetchall() return [User(*row) for row in rows]
service層：ビジネスロジックを書く
service/user_service.py
「ユーザー登録時に重複チェックをする」といったアプリケーション固有のルールをここに書きます。UserRepository という抽象にだけ依存しているため、裏側がSQLiteかインメモリかをまったく意識しません。これがリポジトリパターンの最大の恩恵です。
from domain.user import User from domain.user_repository import UserRepository class UserService: def __init__(self, repo: UserRepository): self.repo = repo def register(self, user_id: int, name: str, email: str) -> None: if self.repo.find_by_id(user_id): raise ValueError(f"ユーザー {user_id} は既に存在します") self.repo.save(User(id=user_id, name=name, email=email)) def get_user(self, user_id: int) -> User: user = self.repo.find_by_id(user_id) if not user: raise ValueError(f"ユーザー {user_id} が見つかりません") return user
4. 組み立てる — main.py
どの実装を使うかを決める唯一の場所です。repo = ... の1行を差し替えるだけで動作環境を切り替えられます。UserService 自身はこの選択を知らなくてよいため、ビジネスロジックに一切手を加えずに環境を変えられます。
from infrastructure.in_memory_user_repository import InMemoryUserRepository from infrastructure.sqlite_user_repository import SQLiteUserRepository from service.user_service import UserService # テスト・開発時 repo = InMemoryUserRepository() # 本番時（1行差し替えるだけ） # repo = SQLiteUserRepository("prod.db") service = UserService(repo=repo) service.register(1, "田中太郎", "tanaka@example.com") service.register(2, "鈴木花子", "suzuki@example.com") user = service.get_user(1) print(user) all_users = repo.find_all() print(all_users)
main.py だけが全体を知っています。service 層も infrastructure 層も、互いの存在を直接知ることなく、domain 層の抽象を介して連携しています。
5. まとめ
リポジトリパターンが解決する3つの課題と、その手段を整理します。
| 課題 | 解決手段 |
|---|---|
| DBを変えるとビジネスロジックまで変更が必要 | 抽象リポジトリに依存させることで、実装の差し替えをビジネスロジックから隔離する |
| テストに実DBが必要で遅い・不安定 | InMemoryUserRepository に差し替えるだけでDBなしテストが可能になる |
| SQLが各所に散らばり保守しにくい | データアクセスのコードをリポジトリクラスに集約する |
設計の核心は依存の方向を一方向に保つことです。domain 層は何も知らなくてよく、infrastructure 層だけが domain 層を知る。この原則を守ることで、アプリケーションの内側（ビジネスロジック）を外側の都合（DBの種類、通信方式）から守り続けることができます。
補足：Webアプリに発展させたら main.py はどう変わる？
今回のサンプルでは main.py がインスタンスの組み立てとサービスの呼び出しを兼ねていますが、Webアプリに発展させると役割が分離されます。
変更前（今回のサンプル）
my_app/ ├── domain/ ├── infrastructure/ ├── service/ └── main.py ← サービスの直接呼び出しを兼ねている
変更後（Webアプリ）
my_app/ ├── domain/ ├── infrastructure/ ├── service/ ├── controller/ │ └── user_controller.py ← HTTPリクエストを受け取りServiceを呼ぶ ├── dependencies.py ← DIの組み立てを集約する └── main.py ← サーバー起動のみ
コントローラーはHTTPリクエストを受け取り、サービスに処理を委譲して、レスポンスを返す専用の層です。main.py はサーバーの起動だけに専念し、リクエストの処理には関与しなくなります。
DIの組み立ては dependencies.py に集約します。エンドポイントが増えてもDB接続の設定変更がここ1か所に収まります。FastAPIでは Depends() を使った依存性注入によって、リクエストのたびに get_user_service() が呼び出され、返ってきた UserService インスタンスがコントローラーの引数に自動的に注入されます。
dependencies.py
def get_user_service() -> UserService: repo = SQLiteUserRepository("prod.db") return UserService(repo=repo)
controller/user_controller.py（FastAPIの例）
@router.post("/users") def create_user(body: CreateUserRequest, service: UserService = Depends(get_user_service)): service.register(body.id, body.name, body.email)
main.py
app = FastAPI() app.include_router(user_controller.router)
domain / service / infrastructure の3層はそのまま変わりません。Webアプリへの発展は、main.py にコントローラー層を加えるだけで済みます。
【Python】PEP 544のProtocolについて
この記事はAIで書いています。
【Python】PEP 544（Protocol）を初心者向けに分かりやすく解説！ダックタイピングを型ヒントでも実現する仕組み
Pythonの型ヒントを勉強していると、「PEP 544」や「Protocol（プロトコル）」という言葉を見かけることがありませんか？
一言でいうと、PEP 544は「Python本来の『ダックタイピング』の柔軟性を、静的型チェックでも安全に使えるようにした仕組み」です。
この記事では、PEP 544が導入された背景と、具体的な使い方を分かりやすく解説します。
1. そもそも「ダックタイピング」とは？
Pythonには、昔からダックタイピング（Duck Typing）という文化があります。
「アヒル（Duck）のように歩き、アヒルのように鳴くなら、それはアヒルだ」
つまり、オブジェクトの「実際のクラス（名前）」が何であるかは重要ではなく、「必要なメソッドや属性を持っているか（何ができるか）」を重視する考え方です。
2. これまでの型ヒントの問題点
しかし、Python 3.5で導入された「型ヒント（Type Hints）」は、基本的に「名目的部分型（Nominal Subtyping）」でした。 これは、クラスの「名前（継承関係）」をベースに型をチェックする仕組みです。
例えば、以下のコードを見てみましょう。
class Dog: def speak(self) -> None: print("Woof!") # Dog型（またはDogを継承したクラス）しか受け付けない def make_it_speak(animal: Dog): animal.speak()
このとき、まったく同じ speak() メソッドを持つ Cat クラスを作ったとします。
class Cat: def speak(self) -> None: print("Meow!") cat = Cat() make_it_speak(cat) # 実行はできるが、型チェック（mypyなど）でエラーになる！
Pythonの実行環境としては問題なく動きます（ダックタイピング）。 しかし、型チェックツール（mypyやPyrightなど）からは「Dog型が必要なのにCat型が渡されている」と怒られてしまいます。
これを解決するためにわざわざ共通の親クラスを作って継承させるのは、Pythonの柔軟性を損なってしまい不便でした。
3. PEP 544の解決策：「構造的部分型」と Protocol
この問題を解決するために、PEP 544で導入されたのが「構造的部分型（Structural Subtyping）」です。 クラスの名前ではなく、クラスの「構造（持っているメソッド）」で型を判断します。
これを実現するために使うのが、typing.Protocol です。
具体的なコード例
from typing import Protocol # 1. 「speakメソッドを持っている」というルール（プロトコル）を定義 class Speaker(Protocol): def speak(self) -> None: ... # 中身は省略（パス）する # 2. 普通にクラスを作る（明示的な継承は不要！） class Dog: def speak(self) -> None: print("Woof!") class Cat: def speak(self) -> None: print("Meow!") # 3. 引数の型に「Protocol」を指定する def make_it_speak(animal: Speaker): animal.speak() # どちらを渡しても、型チェックを無事に通過する！ make_it_speak(Dog()) make_it_speak(Cat())
何が起きているの？
Speakerという「ルール（インターフェース）」を定義しました。DogもCatも、Speakerを継承していません。- しかし、どちらも
speak()メソッドを持っているので、自動的にSpeaker型の一種として認められます。
4. PEP 544（Protocol）を使うメリット
- 明示的な継承（インポート）が不要 他人が作ったライブラリや標準ライブラリのクラスであっても、必要なメソッドさえ揃っていれば型チェックを通せます。
- コードの結合度が下がる 「このクラスに依存する」のではなく「この機能（メソッド）を持つオブジェクトに依存する」という設計ができるため、コードがスッキリします。
- 安全性と柔軟性の両立 Pythonらしい自由で柔軟なダックタイピングの良さを残しつつ、静的型チェックの恩恵（バグの早期発見、エディタ補完）を最大限に受けられます。
まとめ
- PEP 544 は、Pythonに
Protocolを導入した仕様。 - クラスの継承関係ではなく、持っているメソッド（構造）で型をチェックする仕組み。
- Python本来の「ダックタイピング」を型ヒントの世界で安全に実現できる神機能。
モダンなPython開発（特に大規模開発や堅牢なライブラリ開発）では必須の知識なので、ぜひ活用してみてください！
【BigQuery】JSONテクニック。キーが動的な場合はダミーに置換
先日、BigQueryのJSON型データを扱う際に、キーがバラバラで値の抽出処理ができなくて困った。
こちらの記事を参考にして、キーをダミーの文字列に置換して処理する方法でなんとか解決できた。
例えば下記のようなキーにユーザID、値にユーザ名が入っている場合、普通のやり方だとユーザIDを事前に知らないとユーザ名を抽出できない。
ここでユーザIDを一度「KEY」という文字列に置換してから、JSONのパスを「$.KEY」で指定することで、ユーザ名を取り出すことができる。
WITH sample AS (
SELECT JSON '{"8901": "山田 太郎", "8071": "鈴木 次郎"}' AS json_data
)
SELECT
JSON_VALUE(
REGEXP_REPLACE(
TO_JSON_STRING(json_data),
CONCAT('"', key, '"'),
'"__KEY__"'
),
'$.__KEY__'
) AS value
FROM sample,
UNNEST(JSON_KEYS(json_data)) AS key
JSON型のデータを文字列にして、その中のキーをダミーの文字列に変換するという処理は無理矢理感がスゴイけど、こうする他ないこともまあまあるので、知っておくと役に立つかも。
GKE上にAirflowを構築する
概要
TerraformとHelmを使用して、gcloudコマンドやkubectlコマンドを使わずに、Rodel van Rooijen氏のMediumの記事に沿ってGKE上にAirflowクラスタを作成します。
TerraformとHelmのコードは下記のGitHubリポジトリに置いてます。
元の記事とは異なる点がいくつかあるので、ご注意ください。
- Airflow Web UIはインターネットに公開されます。
- KubernetesのPersistent Volumeを手動で作成 (
kubectl apply) する必要はありません。KubernetesのPersistent Volume ClaimによってCompute Diskが自動的に作成されます。詳細はPersistent volumes and dynamic provisioning（永続ボリュームと動的プロビジョニング）を参照してください。 - ログはPodのライフサイクル中のみ利用可能です。ログを永続化したい場合は、Airflow Helm: Manage logs（Airflow Helm：ログの管理）を参照してください。
- Compute Engine DiskはDAGファイルの管理には使用されません。DAGファイルを効率的に管理したい場合は、Airflow Helm: Manage DAGs files（Airflow Helm：DAGファイルの管理）を参照してください。
terraform applyの前に準備すること
terraform applyするには、以下のコマンドラインツールを準備する必要があります。
- gcloud
- docker
- kubectl
- helm
また、下記の点にご注意ください。
- Terraformのコードは、GKEの認証情報を上書きするために
~/.kubeディレクトリを削除します。必要であれば、事前に~/.kubeディレクトリをバックアップしてください。 - Terraformの適用前にGCPプロジェクトが作成済みであり、必要なGCP APIがすべて有効になっている必要があります。
注意事項
下記の点において本番環境レベルのコードではありません。
- デフォルトのVPCネットワークは、カスタムVPCに置き換えるべきです。
- Cloud SQLのパスワードは、Terraformの変数ではなく、Secret Managerなどのより安全な方法で管理すべきです。
- Terraformのstateファイルは、ローカル環境ではなく、GCS (Google Cloud Storage) に保存すべきです。
- Git-syncを使ったDAGファイルの管理の方が便利です。
- ログはPodのライフサイクル中のみ利用可能です。
このAirflow環境は、DAGの開発などに利用できるとは思います。
ハマったところ
Airflow Helm chartには2つの種類があります。公式Helm chartとコミュニティ版です。
それを理解してない状態で調べたので、色々とごちゃごちゃに理解してしまい、時間がかかってしまいました。
このリポジトリでは公式Helm chartを使用してます。
気づき
開発中に気づいたことのメモ。
- Compute Engine DiskはPVC (Persistent Volume Claim) のReadWriteManyアクセスモードをサポートしてない。
standards-rwoStorage Classは、PVC作成直後にCompute Engine Diskを作成しない。
参考資料
Deploying Airflow on GKE using Helm
Workload Identity in GKE with Terraform
Dynamic Provisioning and Storage Classes in Kubernetes
Persistent volumes and dynamic provisioning
Apache Airflow ETL in Google Cloud
Alternative: link Kubernetes ServiceAccounts to IAM
Deploying Airflow on Google Kubernetes Engine with Helm
Deploying Airflow on Google Kubernetes Engine with Helm — Part Two
AirflowのカスタムTrigger
概要
ComposerのAirflowのカスタムTriggerを作成した。
色々と難しかったけど、使えるとリソース消費を抑えられて、コスト抑制できるのでおススメ。
なぜカスタムTriggerを使ったのか
定期的にAPIでBigQueryテーブルの更新状況をチェックして更新されていたら、テーブルに依存する別のテーブルを更新したかった。
AirflowのWorkerでその処理を実行しようとすると、テーブルが更新されるまでCPUとメモリを占有してしまい、お金がかかる。
更新状況チェック対象のテーブルが大量にあるのと、数時間というレベルで待機が必要になる。
そういったユースケースに対して、導入されたのがTriggerプロセス。
AirflowのOperatorでdeferメソッドを実行すると、Triggererプロセスに処理が渡され、非同期で複数の処理をまとめて実行してくれる。
Triggerの処理が完了すると、呼び出し元のOperatorに処理が戻されてタスクが完了する。
Triggererプロセスに処理が渡されている間は、タスクはdefferedステータスになり、Workerで使用されていたリソースは解放されるので、テーブルが更新されるまで待機するといったタスクでもリソース消費を抑えて処理できる。
カスタムTriggerの動かし方
今回はGCP ComposerのAirflowでカスタムTriggerを動かした。
DAGとは違って、ComposerバケットにDAGのPythonファイルを入れても動かない。
TriggererプロセスはAirflowの起動時に読み込まれて実行が開始される。
そのため、PyPIパッケージとしてComposerにインストールする必要がある。
ComposerにカスタムPyPIパッケージをインストールする方法は色々あるが、今回はArtifact Registryの標準リポジトリに生成したPythonパッケージをアップロードして、そこからComposerにインストールした。
その話はまた別途投稿したい。
コード
基本的には下記を参考にすれば、問題なく実装できると思う。
何か不明な点や解決できない問題があれば、メッセージください。
