データエンジニアリング入門 — ETL、データウェアハウス、ストリーミング、データレイク

生データストア (Data Lake)
├── 構造化データ (CSV, Parquet, ORC)
├── 半構造化データ (JSON, XML, Avro)
└── 非構造化データ (画像, ログ, テキスト)

分析用ストア (Data Warehouse)
├── ファクトテーブル (売上, 注文, クリック)
├── ディメンションテーブル (ユーザー, 商品, 日付)
└── 集約テーブル (日別/月別サマリー)

Bronze（生データ）
  → ソースからそのまま取り込んだデータ
  → 最小限の変換のみ適用

Silver（精製データ）
  → 重複排除、型変換、バリデーション完了
  → ビジネスロジック適用前の状態

Gold（ビジネスデータ）
  → 集約、結合、ビジネスルール適用済み
  → ダッシュボードやMLモデルから直接利用

ソース → [抽出] → [変換] → [ロード] → ウェアハウス

ソース → [抽出] → [ロード] → ウェアハウス → [変換]

-- dbtモデル例: 日別売上集計
-- models/marts/daily_revenue.sql

WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
),

payments AS (
    SELECT * FROM {{ ref('stg_payments') }}
)

SELECT
    o.order_date,
    COUNT(DISTINCT o.order_id) AS total_orders,
    SUM(p.amount) AS total_revenue,
    AVG(p.amount) AS avg_order_value
FROM orders o
JOIN payments p ON o.order_id = p.order_id
WHERE p.status = 'completed'
GROUP BY o.order_date

# PySpark例: 日別売上集計
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum, count

spark = SparkSession.builder \
    .appName("DailyRevenue") \
    .getOrCreate()

# データ読み込み
orders = spark.read.parquet("s3://data-lake/orders/")
payments = spark.read.parquet("s3://data-lake/payments/")

# 変換
daily_revenue = (
    orders
    .join(payments, "order_id")
    .filter(col("status") == "completed")
    .groupBy("order_date")
    .agg(
        count("order_id").alias("total_orders"),
        spark_sum("amount").alias("total_revenue")
    )
    .orderBy("order_date")
)

# 結果の保存
daily_revenue.write \
    .mode("overwrite") \
    .parquet("s3://data-warehouse/daily_revenue/")

# DataFrameを登録してSQL使用
orders.createOrReplaceTempView("orders")
payments.createOrReplaceTempView("payments")

result = spark.sql("""
    SELECT
        o.order_date,
        COUNT(DISTINCT o.order_id) AS total_orders,
        SUM(p.amount) AS total_revenue
    FROM orders o
    JOIN payments p ON o.order_id = p.order_id
    WHERE p.status = 'completed'
    GROUP BY o.order_date
    ORDER BY o.order_date
""")

# Kafka Producer例 (Python)
from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

# 注文イベントの発行
event = {
    "order_id": "ORD-12345",
    "user_id": "USR-678",
    "amount": 45000,
    "timestamp": "2026-04-13T10:30:00Z"
}

producer.send('order-events', value=event)
producer.flush()

# Kafka Consumer例 (Python)
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'order-events',
    bootstrap_servers=['localhost:9092'],
    group_id='order-processing-group',
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    auto_offset_reset='earliest'
)

for message in consumer:
    order = message.value
    print(f"注文処理: {order['order_id']}, 金額: {order['amount']}")

運用DB → [CDC] → Kafka → [Stream Processing] → ウェアハウス
                                               → 検索エンジン
                                               → キャッシュ

{
  "before": null,
  "after": {
    "id": 1001,
    "name": "田中太郎",
    "email": "tanaka@example.com"
  },
  "source": {
    "connector": "postgresql",
    "db": "users_db",
    "table": "users"
  },
  "op": "c",
  "ts_ms": 1681364400000
}

// Flinkストリーム処理例 (Java)
DataStream<OrderEvent> orders = env
    .addSource(new FlinkKafkaConsumer<>(
        "order-events",
        new OrderEventSchema(),
        kafkaProps
    ));

// 5分間のタンブリングウィンドウ集計
DataStream<WindowedRevenue> revenue = orders
    .keyBy(OrderEvent::getCategory)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new RevenueAggregator());

revenue.addSink(new JdbcSink<>(...));

# Airflow DAG例
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.sensors.s3_key_sensor import S3KeySensor
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-engineering',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['team@example.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='daily_revenue_pipeline',
    default_args=default_args,
    description='日別売上データ処理パイプライン',
    schedule_interval='0 2 * * *',  # 毎日午前2時
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['revenue', 'daily'],
) as dag:

    # 1. ソースデータの存在確認
    check_source = S3KeySensor(
        task_id='check_source_data',
        bucket_name='raw-data-bucket',
        bucket_key='orders/{{ ds }}/*.parquet',
        timeout=3600,
        poke_interval=300,
    )

    # 2. データ抽出
    extract = PythonOperator(
        task_id='extract_orders',
        python_callable=extract_orders_from_source,
    )

    # 3. データ変換
    transform = PythonOperator(
        task_id='transform_orders',
        python_callable=transform_and_aggregate,
    )

    # 4. ウェアハウスへのロード
    load = PythonOperator(
        task_id='load_to_warehouse',
        python_callable=load_to_snowflake,
    )

    # 5. データ品質検証
    validate = PythonOperator(
        task_id='validate_data_quality',
        python_callable=run_quality_checks,
    )

    # 依存関係の定義
    check_source >> extract >> transform >> load >> validate

-- ファクトテーブル
CREATE TABLE fact_sales (
    sale_id         BIGINT PRIMARY KEY,
    date_key        INT REFERENCES dim_date(date_key),
    product_key     INT REFERENCES dim_product(product_key),
    customer_key    INT REFERENCES dim_customer(customer_key),
    store_key       INT REFERENCES dim_store(store_key),
    quantity        INT,
    unit_price      DECIMAL(10,2),
    total_amount    DECIMAL(12,2),
    discount_amount DECIMAL(10,2)
);

-- ディメンションテーブル
CREATE TABLE dim_product (
    product_key     INT PRIMARY KEY,
    product_id      VARCHAR(50),
    product_name    VARCHAR(200),
    category        VARCHAR(100),
    subcategory     VARCHAR(100),
    brand           VARCHAR(100),
    unit_cost       DECIMAL(10,2)
);

CREATE TABLE dim_date (
    date_key        INT PRIMARY KEY,
    full_date       DATE,
    year            INT,
    quarter         INT,
    month           INT,
    week            INT,
    day_of_week     VARCHAR(20),
    is_holiday      BOOLEAN
);

import great_expectations as gx

# データコンテキストの作成
context = gx.get_context()

# データソース接続
datasource = context.sources.add_pandas("my_datasource")
data_asset = datasource.add_dataframe_asset(name="orders")

# 期待値（Expectation）の定義
batch = data_asset.build_batch_request(dataframe=df)
validator = context.get_validator(batch_request=batch)

# データ品質ルールの定義
validator.expect_column_values_to_not_be_null("order_id")
validator.expect_column_values_to_be_unique("order_id")
validator.expect_column_values_to_be_between(
    "amount", min_value=0, max_value=1000000
)
validator.expect_column_values_to_be_in_set(
    "status", ["pending", "completed", "cancelled", "refunded"]
)
validator.expect_column_values_to_match_regex(
    "email", r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"
)

# 検証の実行
results = validator.validate()
print(f"成功: {results.success}")

-- Snowflake RBAC例
CREATE ROLE data_analyst;
CREATE ROLE data_engineer;
CREATE ROLE data_admin;

-- アナリストロール: 読み取り専用
GRANT USAGE ON DATABASE analytics_db TO ROLE data_analyst;
GRANT SELECT ON ALL TABLES IN SCHEMA analytics_db.gold TO ROLE data_analyst;

-- エンジニアロール: 読み書き可能
GRANT ALL PRIVILEGES ON DATABASE analytics_db TO ROLE data_engineer;
GRANT ALL PRIVILEGES ON ALL SCHEMAS IN DATABASE analytics_db TO ROLE data_engineer;

-- ユーザーにロールを割り当て
GRANT ROLE data_analyst TO USER analyst_kim;
GRANT ROLE data_engineer TO USER engineer_park;

[ソースシステム]
  ├── 運用DB (PostgreSQL)  ─── Debezium CDC ──┐
  ├── Webイベント (Clickstream) ─── SDK ───────┤
  ├── 外部API ─── Airbyte ────────────────────┤
  └── ファイル (CSV/Excel) ─── Airflow ────────┤
                                                │
[メッセージブローカー]                            │
  └── Apache Kafka <────────────────────────────┘
       ├── リアルタイム経路: Flink → リアルタイムダッシュボード
       └── バッチ経路:      Spark → S3 (Data Lake)
                                    │
[変換レイヤー]                       │
  └── dbt (Silver/Gold) <───────────┘
       │
[ウェアハウス]
  └── Snowflake
       ├── Goldレイヤー → Looker/Tableau ダッシュボード
       └── ML Feature Store → モデル学習

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=10),
    'execution_timeout': timedelta(hours=2),
}

with DAG(
    dag_id='e2e_data_pipeline',
    default_args=default_args,
    schedule_interval='0 3 * * *',
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    # ステージ1: Sparkで生データ処理
    spark_process = SparkSubmitOperator(
        task_id='spark_raw_processing',
        application='s3://scripts/process_raw_data.py',
        conn_id='spark_default',
        conf={
            'spark.executor.memory': '4g',
            'spark.executor.cores': '2',
        },
    )

    # ステージ2: dbtでデータ変換
    dbt_transform = DbtCloudRunJobOperator(
        task_id='dbt_transform',
        job_id=12345,
        check_interval=30,
        timeout=3600,
    )

    # ステージ3: データ品質検証
    quality_check = PythonOperator(
        task_id='data_quality_check',
        python_callable=run_great_expectations_suite,
    )

    # ステージ4: 完了通知
    notify = SlackWebhookOperator(
        task_id='slack_notification',
        slack_webhook_conn_id='slack_webhook',
        message='Daily pipeline completed successfully.',
    )

    spark_process >> dbt_transform >> quality_check >> notify

# Spark Structured Streaming + Kafka
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, window, sum as spark_sum
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

spark = SparkSession.builder \
    .appName("RealTimeRevenue") \
    .getOrCreate()

# Kafkaスキーマ定義
order_schema = StructType([
    StructField("order_id", StringType()),
    StructField("user_id", StringType()),
    StructField("amount", DoubleType()),
    StructField("category", StringType()),
    StructField("timestamp", TimestampType()),
])

# Kafkaからストリーミングデータを読み込み
raw_stream = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "order-events")
    .option("startingOffsets", "latest")
    .load()
)

# JSONパース
orders = (
    raw_stream
    .select(from_json(
        col("value").cast("string"),
        order_schema
    ).alias("data"))
    .select("data.*")
)

# 5分ウィンドウ集計
windowed_revenue = (
    orders
    .withWatermark("timestamp", "10 minutes")
    .groupBy(
        window("timestamp", "5 minutes"),
        "category"
    )
    .agg(spark_sum("amount").alias("revenue"))
)

# 結果をコンソールに出力（本番ではDBやダッシュボードへ）
query = (
    windowed_revenue.writeStream
    .outputMode("update")
    .format("console")
    .option("truncate", False)
    .trigger(processingTime="1 minute")
    .start()
)

query.awaitTermination()