transforming-data

データ変換

最新の変換パターン、フレームワーク、オーケストレーションツールを使用して、生データを分析用アセットに変換します。

目的

最新のデータスタック全体で、データ変換パターンを選択して実装します。SQL（dbt）、Python の DataFrame（pandas、polars、PySpark）、パイプラインのオーケストレーション（Airflow、Dagster、Prefect）を使用して、生データをクリーンで、テスト済みかつドキュメント化された分析用データセットに変換します。

使用するタイミング

次の場合にこのスキルを使用します。

• ETL と ELT の変換パターンを選択する
• dbt モデル（ステージング、中間、データマート）を構築する
• 増分データ読み込みとマージ戦略を実装する
• パフォーマンス向上のために pandas のコードを polars に移行する
• 依存関係と再試行を設定してデータパイプラインをオーケストレーションする
• データ品質テストと検証を追加する
• PySpark で大規模なデータセットを処理する
• 本番環境に対応した変換ワークフローを作成する

クイックスタート：一般的なパターン

dbt の増分モデル

{{
  config(
    materialized='incremental',
    unique_key='order_id'
  )
}}

select order_id, customer_id, order_created_at, sum(revenue) as total_revenue
from {{ ref('int_order_items_joined') }}
group by 1, 2, 3

{% if is_incremental() %}
    where order_created_at > (select max(order_created_at) from {{ this }})
{% endif %}

polars による高性能な変換

import polars as pl

result = (
    pl.scan_csv('large_dataset.csv')
    .filter(pl.col('year') == 2024)
    .with_columns([(pl.col('quantity') * pl.col('price')).alias('revenue')])
    .group_by('region')
    .agg(pl.col('revenue').sum())
    .collect()  # Execute lazy query
)

Airflow データパイプライン

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

with DAG(
    dag_id='daily_sales_pipeline',
    schedule_interval='0 2 * * *',
    default_args={'retries': 2, 'retry_delay': timedelta(minutes=5)},
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    extract = PythonOperator(task_id='extract', python_callable=extract_data)
    transform = PythonOperator(task_id='transform', python_callable=transform_data)
    extract >> transform

選択の指針

ETL と ELT の選択

次の場合は ELT（抽出、ロード、変換）を使用します。

• 最新のクラウドデータウェアハウス（Snowflake、BigQuery、Databricks）を使用している
• 変換ロジックが頻繁に変更される
• チームに SQL アナリストがいる
• データ量が 10GB～1TB 以上（ウェアハウスの並列処理を活用する）

ツール：dbt、Dataform、Snowflake tasks、BigQuery scheduled queries

次の場合は ETL（抽出、変換、ロード）を使用します。

• 規制遵守のため、ロード前のデータ秘匿化（PII/PHI）が必要
• ターゲットシステムの計算能力が不足している
• 変換を即時に行うリアルタイムストリーミングを実施する
• クラウドデータウェアハウスを使用しないレガシーシステムである

ツール：AWS Glue、Azure Data Factory、カスタム Python スクリプト

機密データのクレンジング（ETL）と分析変換（ELT）を組み合わせる場合は、ハイブリッドを使用します。

既定の推奨：コンプライアンスやパフォーマンスの制約から ETL が必要でない限り、dbt を使用した ELT を推奨します。

詳しいパターンについては、references/etl-vs-elt-patterns.md を参照してください。

DataFrame ライブラリの選択

次の場合は pandas を選択します。

• データサイズが 500MB 未満
• プロトタイプ作成または探索的分析を行う
• pandas 専用ライブラリとの互換性が必要

次の場合は polars を選択します。

• データサイズが 500MB～100GB
• パフォーマンスが重要（pandas より 10～100 倍高速）
• メモリ制約のある本番パイプラインで使用する
• クエリ最適化を伴う遅延評価を利用したい

次の場合は PySpark を選択します。

• データサイズが 100GB を超える
• クラスター全体での分散処理が必要
• 既存の Spark インフラストラクチャ（EMR、Databricks）がある

移行パス：pandas → polars（容易で、API も類似）または pandas → PySpark（クラスターが必要）

比較と移行ガイドについては、references/dataframe-comparison.md を参照してください。

オーケストレーションツールの選択

次の場合は Airflow を選択します。

• エンタープライズの本番環境で使用する（大規模運用で実績がある）
• 5,000 以上の統合が必要
• マネージドサービスを利用できる（AWS MWAA、GCP Cloud Composer）

次の場合は Dagster を選択します。

• dbt を多用する（ネイティブの dbt_assets 統合）
• データリネージとアセットベースのワークフローを重視する
• テスト容易性が求められる ML パイプラインを構築する

次の場合は Prefect を選択します。

• 動的なワークフローを使用する（実行時にタスクを生成する）
• クラウドネイティブなアーキテクチャを優先する
• デコレーターを使用する Python らしい API を求める

無難な既定の選択：Dagster や Prefect を必要とする特別な要件がない限り、実績豊富な Airflow を使用します。

詳しいパターンについては、references/orchestration-patterns.md を参照してください。

dbt による SQL 変換

モデルレイヤーの構成

1. ステージングレイヤー（models/staging/）

   • ソーステーブルと 1 対 1 で対応
   • 最小限の変換（名前の変更、型変換、基本的なフィルタリング）
   • ビューまたはエフェメラルとしてマテリアライズ

2. 中間レイヤー（models/intermediate/）

   • ビジネスロジックと複雑な結合
   • エンドユーザーには公開しない
   • 多くの場合、エフェメラル（CTE のみ）

3. データマートレイヤー（models/marts/）

   • レポート用の最終モデル
   • ファクトテーブル（イベント、トランザクション）
   • ディメンションテーブル（顧客、製品）
   • テーブルまたは増分モデルとしてマテリアライズ

dbt のマテリアライゼーションタイプ

ビュー：モデルが参照されるたびにクエリが再実行されます。高速なクエリやステージングレイヤーに使用します。

テーブル：実行のたびに全件リフレッシュされます。頻繁にクエリされるモデルや、計算コストの高い処理に使用します。

インクリメンタル：新規または変更されたレコードのみを処理します。大規模なファクトテーブルやイベントログに使用します。

エフェメラル：CTE のみで、永続化されません。中間計算に使用します。

dbt のテスト

models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: customer_id
        tests:
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: total_revenue
        tests:
          - dbt_utils.accepted_range:
              min_value: 0

dbt のパターンについて詳しくは、以下を参照してください。

• references/dbt-best-practices.md
• references/incremental-strategies.md

Python の DataFrame 変換

pandas による変換

import pandas as pd

df = pd.read_csv('sales.csv')
result = (
    df
    .query('year == 2024')
    .assign(revenue=lambda x: x['quantity'] * x['price'])
    .groupby('region')
    .agg({'revenue': ['sum', 'mean']})
)

polars による変換（10～100 倍高速）

import polars as pl

result = (
    pl.scan_csv('sales.csv')  # Lazy evaluation
    .filter(pl.col('year') == 2024)
    .with_columns([(pl.col('quantity') * pl.col('price')).alias('revenue')])
    .group_by('region')
    .agg([
        pl.col('revenue').sum().alias('revenue_sum'),
        pl.col('revenue').mean().alias('revenue_mean')
    ])
    .collect()  # Execute lazy query
)

主な違い：

• polars は scan_csv()（遅延評価）を使用するのに対し、pandas は read_csv()（即時評価）を使用します
• polars は with_columns() を使用するのに対し、pandas は assign() を使用します
• polars は pl.col() 式を使用するのに対し、pandas は文字列参照を使用します
• polars では、遅延クエリを実行するために collect() が必要です

分散処理向けの PySpark

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("Transform").getOrCreate()
df = spark.read.csv('sales.csv', header=True, inferSchema=True)

result = (
    df
    .filter(F.col('year') == 2024)
    .withColumn('revenue', F.col('quantity') * F.col('price'))
    .groupBy('region')
    .agg(F.sum('revenue').alias('total_revenue'))
)

移行ガイドについては、references/dataframe-comparison.md を参照してください。

パイプラインのオーケストレーション

Airflow DAG の構造

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-engineering',
    'retries': 2,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    dag_id='data_pipeline',
    default_args=default_args,
    schedule_interval='0 2 * * *',  # Daily at 2 AM
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    task1 = PythonOperator(task_id='extract', python_callable=extract_fn)
    task2 = PythonOperator(task_id='transform', python_callable=transform_fn)
    task1 >> task2  # Define dependency

タスクの依存関係パターン

直列：A >> B >> C（順次実行） ファンアウト：A >> [B, C, D]（A の後に並列実行） ファンイン：[A, B, C] >> D（すべての処理が完了するまで D は待機）

Airflow、Dagster、Prefect のパターンについては、references/orchestration-patterns.md を参照してください。

データ品質とテスト

dbt のテスト

ジェネリックテスト（再利用可能）：unique、not_null、accepted_values、relationships

単一テスト（カスタム SQL）：

-- tests/assert_positive_revenue.sql
select * from {{ ref('fct_orders') }}
where total_revenue < 0

Great Expectations

import great_expectations as gx

context = gx.get_context()
suite = context.add_expectation_suite("orders_suite")

suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="total_revenue", min_value=0
    )
)

テストパターンについて詳しくは、references/data-quality-testing.md を参照してください。

高度な SQL パターン

分析用のウィンドウ関数：

select
    order_date,
    daily_revenue,
    avg(daily_revenue) over (
        partition by region
        order by order_date
        rows between 6 preceding and current row
    ) as revenue_7d_ma,
    sum(daily_revenue) over (
        partition by region
        order by order_date
    ) as cumulative_revenue
from daily_sales

高度なウィンドウ関数については、references/window-functions-guide.md を参照してください。

本番環境でのベストプラクティス

冪等性

変換を複数回実行しても同じ結果になるようにします。

• インクリメンタルモデルでは merge 文を使用します
• 重複排除ロジックを実装します
• dbt のインクリメンタルモデルでは unique_key を使用します

インクリメンタルロード

{% if is_incremental() %}
    where created_at > (select max(created_at) from {{ this }})
{% endif %}

エラー処理

try:
    result = perform_transformation()
    validate_result(result)
except ValidationError as e:
    log_error(e)
    raise

モニタリング

• タスク失敗時の Airflow のメール／Slack アラートを設定します
• dbt のテスト失敗を監視します
• データの鮮度（SLA）を追跡します
• 行数とデータ品質メトリクスを記録します

ツールの推奨事項

SQL 変換：dbt Core（業界標準、複数のデータウェアハウスに対応、豊富なエコシステム）

pip install dbt-core dbt-snowflake

Python DataFrame：polars（pandas より 10～100 倍高速、マルチスレッド、遅延評価）

pip install polars

オーケストレーション：Apache Airflow（大規模環境で実績あり、5,000 以上の連携）

pip install apache-airflow

例

動作するサンプル：

• examples/python/pandas-basics.py - pandas による変換
• examples/python/polars-migration.py - pandas から polars への移行
• examples/python/pyspark-transformations.py - PySpark の操作
• examples/python/airflow-data-pipeline.py - Airflow DAG の完全な例
• examples/sql/dbt-staging-model.sql - dbt のステージングレイヤー
• examples/sql/dbt-intermediate-model.sql - dbt の中間レイヤー
• examples/sql/dbt-incremental-model.sql - インクリメンタルパターン
• examples/sql/window-functions.sql - 高度な SQL

スクリプト

• scripts/generate_dbt_models.py - dbt モデルのボイラープレートを生成
• scripts/benchmark_dataframes.py - pandas と polars のパフォーマンスを比較

関連スキル

データ取り込みのパターンについては、ingesting-data を参照してください。 データの可視化については、visualizing-data を参照してください。 データベース設計については、databases-* スキルを参照してください。 リアルタイムストリーミングについては、streaming-data を参照してください。 データプラットフォームのアーキテクチャについては、ai-data-engineering を参照してください。 パイプラインのモニタリングについては、observability を参照してください。