pyspark-etl

PySpark ETL

このスキルでは、PySpark、Spark SQL、Apache Iceberg を使って、本番環境に対応でき、テスト可能な ETL パイプラインを構築するためのパターンを紹介します。プロジェクト構成、結合やウィンドウ関数のイディオム、安全な累積テーブルのマージパターンを含みます。

PySpark ETL ジョブの構築ワークフロー

1. ジョブクラスのひな形を作成する — SparkSession のライフサイクルを管理し、テスト用にセッションを注入できるクラスを作成して、抽象メソッド run_job を公開します。
2. ファクトリー関数で設定を定義する — 設定はプレーンなデータクラスとして保持し、CLI 引数の解析は別のファクトリー関数で行います。これにより、テストで sys.argv に触れずに設定を構築できます。
3. 共有のパーティション対応リーダーでソースデータを読み込む — 日付フィルター、時間範囲、最新パーティションの検索に汎用リーダーユーティリティを使用し、ビジネスフィルターは ETL クラスに記述します。
4. .transform() でパイプラインを構成する — 名前付きメソッドを連結します（read_source().transform(self.enrich).transform(self.merge_with_existing)）。これにより、run_job は純粋なオーケストレーションに徹します。
5. イディオムに沿って変換を適用する — withColumn の連鎖より select を使い、結合タイプとウィンドウフレームを明示し、UDF ではなくネイティブ関数を使います。
6. スキーマ進化に安全な方法で書き込む — Iceberg テーブルへの書き込み時に .byName() を使い、列の順序に依存しないようにします。
7. 出力を検証する — 書き込みのたびに、主キーの一意性とキー列の NULL 件数を確認します。
8. ローカルでテストする — 小規模な DataFrame を手作業で作成し、ローカルの SparkSession で変換メソッドを単体テストします。

プロジェクト構成

ETL クラスのひな形

from abc import ABC, abstractmethod
import logging
from pyspark.sql import SparkSession

class BaseETL(ABC):
    def __init__(self, config, app_name="ETL Job", spark_session=None):
        self.spark = spark_session or SparkSession.builder.appName(app_name).getOrCreate()
        self.config = config
        self.logger = logging.getLogger(self.__class__.__name__)

    @abstractmethod
    def run_job(self): ...

    def stop(self):
        self.spark.stop()

ファクトリー関数による設定

データクラスは純粋なデータとして保持し、CLI の解析は単独のファクトリー関数に記述します。これにより、テストで設定を簡単に構築できます。

import argparse
from dataclasses import dataclass

@dataclass
class MyConfig:
    read_date: int = 20260101

def create_config() -> MyConfig:
    parser = argparse.ArgumentParser()
    parser.add_argument("--read_date", type=int, default=20260101)
    args = parser.parse_args()
    return MyConfig(read_date=args.read_date)

パーティション対応の共有リーダー

パーティション処理用の汎用リーダーを 1 つ作成し、ドメイン固有のフィルターは一時的なリーダークラスに埋め込まず、ETL 内に見える形で記述します。

import pyspark.sql.functions as F

class PartitionedReader:
    @staticmethod
    def read_latest(spark, table_name, partition_col):
        row = spark.read.table(table_name).agg(F.max(partition_col)).first()
        if row is None or row[0] is None:
            return spark.createDataFrame([], spark.read.table(table_name).schema)
        return spark.read.table(table_name).filter(F.col(partition_col) == row[0])

    @staticmethod
    def read_by_date(spark, table_name, partition_col, date_value):
        return spark.read.table(table_name).filter(F.col(partition_col) == date_value)

events = PartitionedReader.read_by_date(spark, "catalog.my_table", "event_date", 20260319)
events = events.filter(F.col("event_type").isin("login", "purchase"))

コードスタイル

• 関数を import pyspark.sql.functions as F としてインポートし、df.colA のような属性アクセスではなく、常に F.col() を使います。属性アクセスでは列が特定の DataFrame 変数に結び付けられるため、結合後や再代入後に問題が発生します。
• .filter() または F.when() 内の複雑なブールロジックが 3 つを超える式で構成される場合は、名前付き変数に抽出します。
• withColumn の連鎖より select を優先します。select では出力スキーマを一度に記述できますが、withColumn を呼び出すたびにクエリプランへプロジェクションが追加されます。
• withColumnRenamed ではなく .alias() を使います。
• 1 つの文で連鎖するメソッド呼び出しは 5 個までにします。select／filter の連鎖、withColumn の連鎖、join の連鎖は、操作の種類ごとに分けます。

# BAD — 3 intermediate DataFrames, one projection per call
df = df.withColumn("a", F.col("a").cast("double"))
df = df.withColumn("b", F.upper(F.col("b")))
df = df.withColumn("c", F.lit(1))

# GOOD — one DataFrame, explicit schema contract
df = df.select(
    F.col("a").cast("double"),
    F.upper(F.col("b")).alias("b"),
    F.lit(1).alias("c"),
)

結合

• 常に how= を明示し、デフォルトに依存しないでください。
• 可読性を高めるため、右結合より左結合を優先します。how="right" を使う代わりに、DataFrame の順序を入れ替えます。
• 結合後の曖昧さを解消するには、すべての列を withColumnRenamed で変更するのではなく、DataFrame 全体にエイリアスを設定します。
• 大きなファクトテーブルと結合する際は、小さなディメンションテーブルやルックアップテーブルを F.broadcast() でラップします。特に、Spark がサイズを自動的に推定できないフィルターや変換を適用した後は重要です（spark.sql.autoBroadcastJoinThreshold による自動ブロードキャストは、通常 10MB 未満など、Spark がサイズを把握できるテーブルに限られます）。df.explain() で BroadcastHashJoin と SortMergeJoin のどちらが使われているか確認します。
• 想定外の重複行を隠すために .dropDuplicates() を安易に使わないでください。根本原因を特定します。また、.dropDuplicates() はシャッフルのオーバーヘッドも追加します。

flights = flights.alias("f")
parking = parking.alias("p")
result = flights.join(F.broadcast(parking), "code", how="left").select(
    F.col("f.start_time").alias("flight_start"),
    F.col("p.total_time").alias("parking_total"),
)

ウィンドウ関数

from pyspark.sql import Window as W を使用します。

• フレームを必ず明示します。指定しない場合、F.sum().over(w) は orderBy の有無によって動作が異なります（累積合計または合計）。
• row_number() とフィルター（行を削除して最適な行を残す）と、ウィンドウ上の first()（列を上書きしてすべての行を残す）の違いを把握します。
• F.first()／F.last() には ignorenulls=True を渡します。そうしないと、パーティションの最初の行に null がある場合、パーティション全体の結果が null になります。
• 空の partitionBy() は避けます。すべてのデータが単一のパーティションに集約されます。グローバル集計には代わりに .agg() を使用します。

w = W.partitionBy("key").orderBy("num").rowsBetween(W.unboundedPreceding, W.unboundedFollowing)
df = df.withColumn("version", F.first("version", ignorenulls=True).over(w))

Map と Array の高階関数

• マップのマージで、一方の値を無条件に優先するのではなく、キーごとに競合を解決する必要がある場合（例：タイムスタンプが後のエントリーを保持する場合）は、map_concat ではなく map_zip_with を使用します。
• UDF を使わずにネストされた構造体から値を取り出すには、transform と array_max／array_min を使用します。
• UDF は避けます。まず、Spark の組み込み関数または高階関数がないか確認します。UDF は Catalyst の最適化を妨げ、シリアライズのオーバーヘッドを増やします。

merged = F.map_zip_with(
    new_map, existing_map,
    lambda key, v1, v2: (
        F.when(v1.isNull(), v2)
        .when(v2.isNull(), v1)
        .otherwise(F.when(v1.event_ts >= v2.event_ts, v1).otherwise(v2))
    ),
)

累積テーブル／スナップショットテーブルのパターン

• マージは冪等である必要があります。同じ入力データで再実行しても、重複が発生してはいけません。
• マージは順序に依存しない必要があります。古いデータをバックフィルしても、新しいデータが上書きされてはいけません。競合は明示的な基準（イベントのタイムスタンプ、バージョン番号、パーティション日付）で解決し、coalesce の引数順のような位置による優先順位は使用しません。
• 書き込みのたびに監査手順として、主キーの一意性とキー列の null 件数を検証します。

データ品質とパフォーマンス

• 空の列には F.lit(None) を使用します。空文字列や "NA" のようなセンチネル値は使用しません。
• カテゴリのマッピングで、.when() のチェーンにおける包括的な処理として .otherwise() を使用するのは避けます。マッピングされていない値は「Other」に暗黙的にまとめず、null として検出できるようにします。
• 本番コードに .show()、.collect()、.printSchema() を残してはいけません。これらはデータをすべて実体化するか、ドライバーのオーバーヘッドを増やします。行数のログ記録や DAG の分岐前に意図的にデータを実体化する用途であれば、.count() は問題ありません。
• .persist() は、DataFrame が後続の複数のアクションで参照される場合にのみ使用します。ストレージレベルは慎重に選択します。MEMORY_AND_DISK（安全なデフォルト）、MEMORY_ONLY（最速ですが、追い出されたデータの再計算が発生する可能性があります）、DISK_ONLY（メモリに収まらない大きさの DataFrame 向け）。

Iceberg の書き込みパターン

• 書き込み時には .byName() を使用し、Spark が列を位置ではなく名前で対応付けるようにします。これにより、スキーマ進化にも安全に対応できます。

df.write.byName().mode("overwrite").insertInto("catalog.my_table")

• テーブル全体をスキャンする代わりに、__partitions Iceberg メタデータテーブルを使用して最新のスナップショットを確認します。

partition_df = spark.read.table("catalog.my_table__partitions").select(
    "partition.partition_date", "partition.partition_hour"
)
max_partition = partition_df.orderBy(
    F.col("partition_date").desc(), F.col("partition_hour").desc()
).first()
if max_partition is None:
    raise ValueError("No partitions found in catalog.my_table")

• write.distribution-mode は慎重に選択します。"none"（最速で、再シャッフルなし。ファイルサイズは上流のパーティション分割に依存）、"hash"（パーティションキーでシャッフルし、ファイルサイズを均等化）、"range"（書き込み前にソート。スキャン性能は最も高くなりますが、コストも最も高くなります）。