🔖 キーワード索引

「データエンジニアリング (data engineering)」は信頼性の高いデータパイプラインを設計・構築・運用する工学分野。 ETL/ELT・データレイク・データウェアハウス・データレイクハウス・データメッシュなどのアーキテクチャと、 dbt・Airflow・Spark などのツール群を活用する。 本ページでは Modern Data Stack・バッチ vs ストリーミング・スキーマ進化・データ品質テスト・観測可能性を整理する。

これらのキーワードは「データを集める → 流す → 保つ → 信頼を担保する」というデータエンジニアリングの中核責務を構成する。

🔖 キーワード索引（チップから該当箇所へジャンプ）

各リンクをクリックすると、 該当セクションへジャンプします：

🎨 直感で掴む — データエンジニアリングの正体

データエンジニアリングは 「料理の下ごしらえ」 です。 シェフ（モデル）の腕がいくら良くても、 食材（データ）が泥だらけだったり、 単位がバラバラだったりすれば、 美味しい料理（精度の良い予測）は作れません。 SSDSE-B-2026 を例にとっても、 「都道府県別人口」と「市区町村別所得」を結合するときに、 主キーの粒度・年次・コード体系（JIS コード）の不整合を整える作業が、 すべての分析の土台になります。

具体的には ETL（Extract → Transform → Load） のサイクルでデータを流通させます。 SSDSE-B-2026 → 型変換 → 結合 → 欠損補完 → 標準化 → モデル投入、 という一連の道筋を Pipeline オブジェクト として固定化すれば、 同じ前処理を train と test に一貫して適用できるため データリーク防止 にも直結します。

🎨 補講：バッチ vs ストリーミング — SSDSE で考える選択軸

データエンジニアの設計上の最大の分岐が バッチ vs ストリーミング。 SSDSE-B-2026 のような「年次更新の公的統計」はバッチが最適。 一方、 EC サイトのクリックログや IoT センサー値はストリーミングが必要。 選択基準を整理します。

📊 比較表

📐 数式：ストリーミングのウィンドウ集計

$$\text{TumblingWindow}(t) = \sum_{i: t - W \le t_i < t} x_i \quad (W: \text{window size})$$

🔬 数式を言葉で読み解く

• $W$：ウィンドウ幅（例：5 分）。 過去 $W$ 秒間のイベントを集計。
• $x_i$：時刻 $t_i$ に到着したイベントの値。
• Tumbling：重ならない固定ウィンドウ（00:00-00:05, 00:05-00:10）。
• Sliding：重なるウィンドウ（過去 5 分を毎分計算）。
• SSDSE は年次データなので、 これは「年集計」に相当。 バッチで十分。

🐍 Python 実装 12：SSDSE 年次データをミニ「バッチパイプライン」化

🎯 このコードでやること：SSDSE-B-2026 を year でグループ化し、 全国合計を年次集計するシンプルなバッチを書く。 これがストリーミングなら「リアルタイム集計」になる。

📥 入力データ：SSDSE-B-2026.csv 全行。

1
2
3
4
5
6
7
8
9
10
11
12

import pandas as pd

df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])
df = df.rename(columns={'SSDSE-B-2026': 'year'})

# 「年次ウィンドウ」での総人口集計（バッチ）
agg = df.groupby('year')['A1101'].sum().sort_index(ascending=False)
print("日本全国 総人口（年次推移、 直近 5 年）")
print(agg.head(5).apply(lambda v: f"{v:,.0f}"))

print(f"\n2023 → 2018 の変動: {agg.loc[2023] - agg.loc[2018]:,.0f} 人")
print(f"年率変化     : {(agg.loc[2023]/agg.loc[2018])**(1/5) - 1:+.4%}")

📤 実行すると次の出力が得られる：

💬 結果の読み方：日本の総人口は 2018→2023 で約 240 万人減（年率 -0.38%）。 これがリアルタイムストリーミングだったら「秒ごと」に更新できますが、 人口統計に秒粒度は不要 → バッチが正解。 「データの意思決定粒度に合わせて頻度を選ぶ」のが正しい設計です。

⚠️ バッチ vs ストリーミング選択の落とし穴

• 「リアルタイム」が目的化：実は日次で十分なのに Kafka を立てて運用負荷だけ上がる。 まず「意思決定の頻度」を聞く。
• Lambda アーキテクチャの二重管理：バッチとストリームで同じロジックを 2 回書く → バグの温床。 Kappa（ストリーム単一）か dbt incremental で一本化。
• ストリーミングで遅延データを忘れる：watermark や late arrival 設計がないと「過去の集計値」がずれる。 Flink/Beam で event-time + watermark を必ず設計。
• バッチの「ジョブ依存地獄」：A→B→C→D の鎖が長すぎて夜中まで終わらない。 DAG を浅く（並列度を上げる）設計に。

📚 まとめ：データエンジニアリングの 3 行サマリ

1. 原則：取得 → 変換 → 配信 → 監視 を、 SLI/SLO で定量管理。
2. 選択軸：行 vs 列、 バッチ vs ストリーム、 ETL vs ELT、 full vs incremental。
3. 姿勢：コード半分 + ガバナンス半分。 「この数字どこから？」に即答できる pipeline を作る。

本ページの全 12 個の Python 実装は、 すべてあなたのローカル PC で data/raw/SSDSE-B-2026.csv 1 ファイルから動きます。 まずは 実装 1（completeness 測定）から始め、 最後の 実装 12（年次バッチ集計）まで自分で動かしてみてください。 SSDSE は小さいですが、 BigQuery で 1TB のデータを扱うときも、 設計原則は同じです。

🎨 まとめの一言：データエンジニアリングは「裏方」だが「主役」

BI ダッシュボードや ML モデルが脚光を浴びがちですが、 その裏で「データが正しく・速く・安く・継続的に流れている」状態を作るのがデータエンジニアの仕事です。 派手ではないが、 ここが崩れるとすべてが崩れる——まさに インフラの役目。

本ページでは SSDSE-B-2026 を題材に 14 個の Python 実装、 SLI/SLO 等の数式、 MDS 全レイヤ、 ガバナンス・リネージ、 バッチ vs ストリーミング までを 1 ファイルで学べる構成にしました。 知識として読むだけでなく、 ぜひ自分の手で 1 本ずつ動かしてみてください。 動かすと、 数字が見え、 設計判断の根拠が体感できるはずです。

最終目標は「3 年後の自分でも 1 時間で原因究明できる pipeline」を作れること。 リネージ・カタログ・SLO・契約・テスト、 これらすべてが将来の自分（や仲間）への贈り物です。

🛠 実践編：ETL vs ELT / Airflow / dbt を SSDSE-B-2026 で動かす

1. ETL と ELT の違い — どちらを選ぶか

SSDSE-B-2026 のような 564 行 × 112 列 の小規模データなら ETL（Extract → Transform → Load）で十分。 一方、 e-Stat 全体（数千万行）を扱う場合は ELT（Extract → Load → Transform）が定石。 ロード先（BigQuery / Snowflake / DuckDB）の計算資源で変換するほうが速い。

2. Airflow — DAG で依存を可視化

🎯 このコードでやること：SSDSE-B-2026 を「取得 → 検証 → 集計 → ロード」する 4 段 DAG を Airflow で書き、 依存関係を Python 表記で示す。

📥 入力例（タスク名と依存）:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

import os
os.makedirs('data', exist_ok=True)   # 書き出し先を先に作る
import os
os.makedirs('data/stg', exist_ok=True)  # 保存先のフォルダを作っておく
os.makedirs('data/mart', exist_ok=True)  # 保存先のフォルダを作っておく

import pyarrow  # parquet の読み書きに必要（ブラウザには無い）
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import pandas as pd, duckdb

def extract():
    df = pd.read_csv('data/raw/SSDSE-B-2026.csv', skiprows=[1], encoding='cp932')
    df.to_parquet('data/stg/ssdse_b_2026.parquet')
    return len(df)

def validate():
    df = pd.read_parquet('data/stg/ssdse_b_2026.parquet')
    assert df['Prefecture'].nunique() == 47, '都道府県数が 47 でない'
    assert df['A1101'].notna().all(), 'A1101 に NA'
    return 'ok'

def aggregate():
    df = pd.read_parquet('data/stg/ssdse_b_2026.parquet')
    g = df.groupby('Prefecture')['A1101'].sum().reset_index()
    g.to_parquet('data/mart/pop_by_pref.parquet')
    return len(g)

def load():
    con = duckdb.connect('data/mart.duckdb')
    con.execute("CREATE OR REPLACE TABLE pop AS SELECT * FROM 'data/mart/pop_by_pref.parquet'")
    return con.execute('SELECT COUNT(*) FROM pop').fetchone()[0]

with DAG('ssdse_b_etl', start_date=datetime(2026, 5, 1), schedule='@monthly', catchup=False) as dag:
    t1 = PythonOperator(task_id='extract', python_callable=extract)
    t2 = PythonOperator(task_id='validate', python_callable=validate)
    t3 = PythonOperator(task_id='aggregate', python_callable=aggregate)
    t4 = PythonOperator(task_id='load', python_callable=load)
    t1 >> t2 >> t3 >> t4

📤 実行例（airflow tasks test ssdse_b_etl extract 2026-05-01）:

💬 >> 演算子で依存を 1 行で書けるのが Airflow の強み。 失敗したタスクから再実行できる（idempotent な処理にしておくこと）。 SSDSE-B のように月次更新がない場合でも、 schedule=@monthly で源データの更新検知 + 通知だけ走らせるのが定石。

3. dbt — SQL 変換のテスト・ドキュメント・lineage を自動生成

dbt を使うと、 SQL ファイルに {{ ref('staging_ssdse_b') }} と書くだけで 依存 DAG が自動生成される。 さらに schema.yml に「not_null / unique / accepted_values」と書くだけでテストが走る。 SSDSE-B のような明確に正規化されたデータは dbt の練習素材として最適。

4. Prefect / dlt — モダンな代替

• Prefect: Airflow より軽量。 @flow / @task デコレータで Python 関数をそのまま DAG 化できる。 SSDSE 程度の規模なら Airflow 構築コストより Prefect Cloud の無料枠が楽。
• dlt (data load tool): ソースから DWH まで「型推論 + スキーマ進化」を自動でやる。 e-Stat API → DuckDB を pipeline.run(source) 1 行で書ける。
• 選択指針: 「Airflow は規模が出てから、 まず Prefect + dlt で MVP」が 2026 年現在の合理的初手。

📌 SSDSE-B-2026 のような「年 1 回更新・1 年あたり 47 行（累計 564 行）」の小規模データに Airflow を立てるのは過剰。 まず Python script + cron + Slack 通知で MVP を作り、 「複数 DAG が絡む」段階で初めて Airflow / Prefect を導入するのが現実的。

🎨 概念図で押さえる

データエンジニアリングは「収集 → 蓄積 → 加工 → 配信」のパイプラインを安定運用する分野。 ここでは「ETL vs ELT」「Lambda / Kappa アーキテクチャ」「データレイクハウス階層」を概念図で押さえる。

ETL (上) と ELT (下) の処理順序ETL: Extract → Transform → LoadSource変換 (Spark等)DWH (整形済)ELT: Extract → Load → TransformSourceLake (生データ)変換 (dbt等)→ 整形済→ 生も保持'>

Lambda アーキテクチャ (バッチ + ストリーム 二重化)SourceBatch Layer日次 Spark / HadoopSpeed Layerリアルタイム KafkaServing Layer統合ビューバッチは正確だが遅い、 ストリームは速いが概算。 両方を統合して堅牢性 + 即時性を両立'>

メダリオン階層 (Bronze → Silver → Gold)🥉 Bronze生データSSDSE-B 原 CSVスキーマほぼ無し🥈 Silver清浄化済tidy + 型統一欠損補完・重複削除🥇 Gold業務集計済KPI / ML特徴量BI / ダッシュボード'>

📐 定義・数式 — 主要な変換の数学

代表的な前処理は以下のとおり数学的に厳密に書けます。

SSDSE-B-2026 で「総人口（A1101）」を z スコア化することは、 K-means や SVM のような距離・勾配ベース手法で 必須 です。 数式に従って $\mu$ と $\sigma$ を train で計算、 同じ値を test に適用するのが鉄則です。

🔬 数式を言葉で読み解く — 記号 → 意味

数式の意味は「平均から何 SD 離れているか（z）」「最小最大の間でどの位置か（Min-Max）」「中央値と IQR で測ったロバストな位置（Robust）」。 用途で使い分けます。

🔬 深掘り：データエンジニアリングの全体像を SSDSE-B-2026 で解剖する

データエンジニアリングは「データを正しく・速く・安く・継続的に流す」職能です。 ここでは公的データ SSDSE-B-2026（47 都道府県 × 12 年 × 100 列以上）を題材に、 取得 → 変換 → 配信 → 監視 の各工程を、 実コードと実数値で具体化します。 単に Airflow を紹介するのではなく、 「なぜその設計か」を式と数値で説明します。

🎨 直感：データエンジニアリング = 「配管 + 検針メーター + 警報装置」

水道に例えると、 取水（収集）→ 浄水（クレンジング）→ 配水（DWH/Lake）→ 蛇口（BI/ML）まで、 すべての配管を保守する仕事です。 水質が悪化したら警報（データ品質アラート）が鳴り、 漏水（欠損・重複）が見つかれば修理します。 単に「ETL を書く人」ではなく、 SLO（Service Level Objective）を守る運用者でもあります。

📐 数式：SLI と SLO で品質を定量化する

データパイプラインの品質は感覚ではなく、 SLI（Service Level Indicator、 サービス指標）と SLO（目標値）で測ります。 代表的な指標は freshness（鮮度）、 completeness（完全性）、 accuracy（正確性）です。

例えば完全性 SLI は次の式で定義できます：

$$\text{Completeness} = 1 - \dfrac{N_{\text{missing}}}{N_{\text{total}}}, \quad \text{Freshness} = T_{\text{now}} - T_{\text{last\_update}}$$

🔬 数式を言葉で読み解く

• Completeness（完全性）：全セル数 $N_{\text{total}}$ のうち、 欠損していないセルの割合。 1.0 に近いほど良い。 SLO 例：「0.99 以上を維持」。
• Freshness（鮮度）：今この瞬間と、 最後にデータが更新された時刻の差。 小さいほど新鮮。 SLO 例：「24 時間以内」。
• $N_{\text{missing}}$：欠損セル数。 $N_{\text{total}}$：全セル数 = 行数 × 列数。
• SLI は「測る」だけ、 SLO は「守るべき目標」。 SLO を下回ったらアラート → On-call 担当が対応、 という運用に直結します。

🧮 実値で計算：SSDSE-B-2026 の品質指標を測る

SSDSE-B-2026 を読み込んで、 上記 SLI を実際に計算してみます。 概算：

• 行数：47 都道府県 × 12 年（2012〜2023）＝ 564 行
• 列数：データ列 109 列（年・コード・都道府県名を除く）
• 総セル数：$N_{\text{total}} = 564 \times 109 = 61{,}476$ セル

実際に NaN を数えると、 SSDSE-B-2026 は欠損ゼロですが、 仮に民間ログのように欠損が 600 セルあったとすると：

SLO「0.99 以上」をギリギリ満たす水準です。 もし 0.99 を下回ったら、 欠損補完ロジック追加 or 過去データの再取得を発動します。

🐍 Python 実装 1：SSDSE-B-2026 を取り込み、 完全性を測る

🎯 このコードでやること：SSDSE-B-2026 を読み込み、 行数・列数・欠損セル数・Completeness（SLI）を計算する。

📥 入力データ（SSDSE-B-2026.csv の先頭 3 行、 cp932）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import pandas as pd

# 1. Ingestion: SSDSE-B-2026 を読み込む（cp932, 2 行目はラベル行なので skip）
df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])

# 2. データ列のみ抽出（年度・地域コード・都道府県名は除く）
data_cols = df.columns[3:]

# 3. SLI を計算
n_rows, n_cols = df.shape[0], len(data_cols)
n_total = n_rows * n_cols
n_missing = df[data_cols].isna().sum().sum()
completeness = 1 - n_missing / n_total
print(f"rows={n_rows}, cols={n_cols}, total={n_total}, missing={n_missing}, completeness={completeness:.4f}")

📤 実行すると次の出力が得られる：

💬 結果の読み方：SSDSE-B-2026 は欠損ゼロの高品質データ（completeness = 1.000）。 SLO「0.99 以上」を余裕でクリア。 公的統計だけあって極めてクリーンです。 もし民間ログデータなら 0.85 程度に落ち込むことも多く、 SLI 監視の意義が出ます。

🐍 Python 実装 2：ETL — wide 形式を long 形式に変換

🎯 このコードでやること：SSDSE-B-2026 の「年度 × 都道府県 × 指標」wide 形式を、 BI/分析向きの long 形式（縦持ち）に melt で変換する。

📥 入力データ（wide 形式・df.head() 結果）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import pandas as pd

df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])
df = df.rename(columns={'SSDSE-B-2026': 'year', 'Code': 'code', 'Prefecture': 'pref'})

# wide → long: 1 行 = (year, pref, metric, value)
long_df = df.melt(
    id_vars=['year', 'code', 'pref'],
    var_name='metric',
    value_name='value'
)

print(long_df.shape)
print(long_df.head(3))

📤 実行すると次の出力が得られる：

💬 結果の読み方：行数が 564 → 61,476 に増えました（109 倍）。 long 形式は「BI ツールが集計しやすい」「新しい指標が増えても列追加不要」というメリットがあります。 一方ファイルサイズは大きくなるため、 Parquet 等の列指向圧縮と組み合わせるのが定石です。

🐍 Python 実装 3：Data Quality チェック — 整合性アサーション

🎯 このコードでやること：SSDSE-B-2026 で「総人口 = 男性人口 + 女性人口」が成り立つかをチェックし、 違反行を表示する（整合性 DQ）。

📥 入力データ：A1101（総人口）, A110101（男）, A110102（女）の 3 列を使用。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import pandas as pd

df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])
df = df.rename(columns={'SSDSE-B-2026': 'year', 'Prefecture': 'pref'})

# 整合性チェック: A1101 (総人口) == A110101 (男) + A110102 (女) ?
df['sum_mf'] = df['A110101'] + df['A110102']
df['diff'] = df['A1101'] - df['sum_mf']

# 1000 人以上の乖離を異常としてフラグ
bad = df[df['diff'].abs() > 1000][['year', 'pref', 'A1101', 'sum_mf', 'diff']]
print(f"violations: {len(bad)} rows")
print(f"最大の乖離: {df['diff'].abs().max()} 人（許容 1000 人）")
if bad.empty:
    print("整合性チェック OK — 総人口 = 男 + 女 がすべての行で成り立っています")
else:
    print(bad.head(5))

📤 実行すると次の出力が得られる：

💬 結果の読み方：違反ゼロ。 SSDSE は四捨五入があるため小さな差はあるものの、 ±1000 人の閾値内で完全整合。 民間データ（行動ログ等）ではこうしたチェックで 1〜10% 程度の不整合が見つかることが多く、 上流バグの早期検知に役立ちます。

🐍 Python 実装 4：DWH 配信 — SQLite に書き出して BI 連携

🎯 このコードでやること：long 形式の SSDSE データを SQLite データベースに投入し、 BI/分析クエリから SQL でアクセス可能にする。

📥 入力データ：直前の long_df（行 61,476）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

import sqlite3
import pandas as pd

df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])
df = df.rename(columns={'SSDSE-B-2026': 'year', 'Code': 'code', 'Prefecture': 'pref'})
long_df = df.melt(id_vars=['year', 'code', 'pref'], var_name='metric', value_name='value')

# DWH 役の SQLite に投入
conn = sqlite3.connect('warehouse.db')
long_df.to_sql('ssdse_b_long', conn, if_exists='replace', index=False)

# 分析クエリ: 2023 年の人口 TOP5
q = """SELECT pref, value AS population FROM ssdse_b_long
       WHERE year = 2023 AND metric = 'A1101'
       ORDER BY value DESC LIMIT 5"""
print(pd.read_sql(q, conn))
conn.close()

📤 実行すると次の出力が得られる：

💬 結果の読み方：BI ツールから「2023 年人口 TOP5」がワンクエリで出ました。 SQLite は単一ファイルなので個人検証向き。 本番では BigQuery / Snowflake / Redshift など列指向 DWH を選びます。 ここまでが「収集 → 変換 → 配信」の最小エンドツーエンド。

⚠️ 落とし穴（深掘り版）

• Schema drift（列が増減する）：SSDSE は年版で列が変わる。 ETL を「列名固定」で書くと翌年に死ぬ。 → 列を melt で動的に長持ち化、 メタデータ駆動で吸収する。
• 文字コードの罠：cp932 / Shift_JIS / UTF-8 / BOM 付き UTF-8 が混在しがち。 pd.read_csv(..., encoding='cp932') を明示し、 失敗時はログに残す。
• 「動いてる」≠「正しい」：パイプラインが緑色でも、 重複・欠損・遅延は静かに進行する。 SLI（freshness/completeness/accuracy）を常時監視し、 SLO 違反でアラート発火する設計が必須。
• コスト爆発：BigQuery で SELECT * を雑に流すと月数十万円飛ぶ。 列指向 DWH では「必要な列・パーティション・clustering」を意識。 dry-run で課金見積もりを取る習慣を。
• Lineage の欠落：「この列、 どこから来たの？」に答えられないと、 障害時に追跡不能。 dbt / OpenLineage / Datahub でデータ系譜を可視化する。
• idempotency（冪等性）の不在：同じ ETL を二度流したら二倍になる、 はバグ。 主キー + MERGE/UPSERT or 日次パーティション置換で冪等にする。
• テストがないパイプライン：データテスト（dbt tests, Great Expectations）を持たない pipeline は「障害が出てから直す」運用になる。 schema, not_null, unique, ranges を最低限カバー。

🌐 関連手法・派生

• ETL vs ELT：従来は Extract → Transform → Load（Transform を中間で）、 現代の DWH は Load してから SQL/dbt で Transform する ELT が主流。
• Data Lake / Lakehouse：S3/GCS にそのまま置く Lake、 さらに ACID/メタ管理を載せた Lakehouse（Delta Lake、 Iceberg、 Hudi）。
• Workflow Orchestration：Airflow / Dagster / Prefect で DAG を組み、 依存関係・リトライ・SLA を管理。
• Streaming：Kafka + Flink/Spark Streaming でリアルタイム処理。 バッチ ETL（Airflow）の対極。
• Data Contracts：上流（プロデューサ）と下流（コンシューマ）でスキーマ・SLA を契約化する近年のトレンド。
• Reverse ETL：DWH の結果を Salesforce / HubSpot 等の SaaS に戻す逆方向の配信。

🗺 概念マップ：データエンジニアリングの全工程

🔗 関連用語（前提・並列・発展）

🔬 補講：ストレージ形式・パーティション・スキーマ管理

データエンジニアリングの 7 割は「ストレージ設計」で性能とコストが決まります。 ここでは SSDSE-B-2026 を題材に、 行指向 vs 列指向、 パーティション設計、 スキーマ進化の実例を、 具体的な数値と Python で示します。

📊 行指向 vs 列指向：何が違うのか

SSDSE-B-2026 のような「都道府県 × 年 × 指標」の集計データは、 ほぼ間違いなく列指向（Parquet）で持つべきです。 「2023 年の総人口列だけ欲しい」というクエリで、 ファイル全体を読まなくて済むからです。

📐 パーティション設計の数式：プルーニング率

$$\text{Pruning Ratio} = 1 - \dfrac{\text{Scanned Partitions}}{\text{Total Partitions}}$$

🔬 数式を言葉で読み解く

• Scanned Partitions：クエリで実際に読み込まれたパーティション数。
• Total Partitions：テーブル全体のパーティション数。
• Pruning Ratio：1.0 に近いほど「クエリが要らない大半をスキップした」状態。 BigQuery 等では課金にも直結。
• SSDSE 例：year でパーティション分割 → 「WHERE year = 2023」で 12 年分のうち 1 年だけスキャン。 Pruning Ratio = 1 - 1/12 ≈ 0.917（91.7% プルーニング）。

🧮 実値で計算：パーティション分割のコスト効果

BigQuery の課金は「スキャンしたバイト数」基準（オンデマンド 6.25 ドル/TB）。 SSDSE-B 想定で計算：

小さなデータでは差は微々たるものですが、 1TB 規模では 月数十万円→数千円に変わります。 「partition 設計はコスト設計」と覚えてください。

🐍 Python 実装 9：Parquet を year でパーティション保存

🎯 このコードでやること：SSDSE-B-2026 を year ごとに分割した Parquet（Hive 形式パーティション）として保存する。

📥 入力データ：SSDSE-B-2026.csv の全 564 行。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import os
os.makedirs('ssdse_part', exist_ok=True)   # 書き出し先を先に作る
import pyarrow  # parquet の読み書きに必要（ブラウザには無い）
import pandas as pd
import os

df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])
df = df.rename(columns={'SSDSE-B-2026': 'year'})

# year をパーティションキーに → ssdse_part/year=2023/part-0.parquet
df.to_parquet('ssdse_part/', partition_cols=['year'], compression='snappy')

# どんなパーティションが作られた？
parts = sorted(os.listdir('ssdse_part'))
print("partitions created:", len(parts))
print("first 3:", parts[:3])
print("last 3:", parts[-3:])

📤 実行すると次の出力が得られる：

💬 結果の読み方：12 個のパーティション（年）が自動で作られた。 「WHERE year=2023」クエリは 1/12 のファイルだけ読めば済む = 約 12 倍速・12 倍安い。 BigQuery / Athena / DuckDB / Spark すべてで同じ仕組み。

🐍 Python 実装 10：スキーマ進化（schema evolution）の検出

🎯 このコードでやること：SSDSE-B-2026 と仮想の「来年版（列追加）」を比較し、 schema drift を検出する。

📥 入力データ：今年の DataFrame と、 列を 1 つ追加した DataFrame。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

import pandas as pd

df_now = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])

# 仮想「来年版」: 既存 + 新列 NEW_METRIC、 一部列が消失
df_next = df_now.copy()
df_next['NEW_METRIC'] = 0
df_next = df_next.drop(columns=['A1101'])   # 仮想削除

added = set(df_next.columns) - set(df_now.columns)
removed = set(df_now.columns) - set(df_next.columns)

print(f"added cols  : {sorted(added)}")
print(f"removed cols: {sorted(removed)}")
print(f"breaking change?: {'YES' if removed else 'no'}")

📤 実行すると次の出力が得られる：

💬 結果の読み方：列追加（NEW_METRIC）は後方互換だが、 列削除（A1101 = 総人口）は 下流の BI/ML が壊れる致命的変更。 CI で検出して PR をブロックする運用を強く推奨。 dbt の contract や Protobuf でも同様の仕組みが使えます。

⚠️ ストレージ設計の落とし穴

• 細かすぎるパーティション：1 日 1 ファイル × 10 年 = 3650 ファイル。 メタデータコストで逆に遅くなる。 1 ファイル 128MB〜1GB を目安に。
• パーティションキーをクエリで使わない：year で分けたのに WHERE pref=... だけ使う → プルーニング効かず全件スキャン。 設計とクエリを揃える。
• 列の型が year で変わる：去年 INT、 今年 STRING → Parquet union 時にエラー。 schema registry や dbt contract で固定化。
• 圧縮の選び間違い：Snappy は速度寄り、 Zstd/Gzip は圧縮率寄り。 ストレージ重視なら Zstd、 クエリ重視なら Snappy。
• Parquet を OLTP に使う：1 行更新が極端に遅い。 OLTP は RDB、 OLAP は Parquet/DWH、 と用途で切り分ける。

ここまでで本ページに登場した Python 実装は 10 本、 すべて SSDSE-B-2026 を入力とし、 ローカルで動きます。 「データエンジニアリングとは何か」を理屈で理解するだけでなく、 ファイルサイズ・スループット・コスト・スキーマの数値を自分の手で測ってこそ、 実務感覚が身につきます。

🔗 補講：データリネージとガバナンス — 「この数字どこから来たの？」に答える

BI ダッシュボードに「東京の総人口 1,408 万人」と出たとき、 ステークホルダーから必ず来る質問が 「この数字、 どこから来たの？」。 これに即答できないと、 データの信頼性は崩壊します。 解は データリネージ（lineage、 系譜）の整備です。

🎨 リネージの直感：「祖先と子孫を辿れる家系図」

BI 上の数字を「子孫」、 元データを「祖先」とすると、 リネージは家系図のようなもの。 BI → SQL view → dbt model → DWH table → ELT → raw csv → 外部 API、 と遡れる必要があります。

🐍 Python 実装 11：簡易リネージグラフの構築

🎯 このコードでやること：SSDSE-B-2026 を起点とする小さなリネージグラフを辞書で表現し、 「BI の指標 → raw csv まで遡る」探索を実装する。

📥 入力データ：パイプラインのノード間依存（辞書）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

lineage = {
    'bi_tokyo_population': ['fact_population_yearly'],
    'fact_population_yearly': ['stg_ssdse_b'],
    'stg_ssdse_b': ['raw_ssdse_b_2026_csv'],
    'raw_ssdse_b_2026_csv': ['external_estat_api'],
    'external_estat_api': [],
}

def trace(node, depth=0):
    print('  ' * depth + '└─ ' + node)
    for parent in lineage.get(node, []):
        trace(parent, depth + 1)

trace('bi_tokyo_population')

# 影響範囲分析: external API が変わったら何に波及する？
downstream = {n for n, ps in lineage.items() if 'external_estat_api' in ps}
print('downstream of external_estat_api:', downstream)

📤 実行すると次の出力が得られる：

💬 結果の読み方：BI の数字を 4 ステップ遡って外部 API まで到達できる。 影響範囲分析（impact analysis）で、 上流変更時の波及先も即座にわかる。 dbt は dbt docs generate で同等のグラフを自動生成します。

🌐 ガバナンスの 3 本柱

• Data Catalog（カタログ）：全テーブル・全列に説明・オーナー・タグを付与（Datahub, OpenMetadata, BigQuery Data Catalog）。
• Access Control（権限）：個人情報列は制限、 集計表は広く公開。 row-level security、 column masking。
• Data Contract（契約）：上流（プロデューサ）と下流（コンシューマ）でスキーマ・SLA を明文化。 破壊的変更を防ぐ。

⚠️ ガバナンスの落とし穴

• カタログを誰も更新しない：作って終わり、 半年で陳腐化。 CI で「新規列に description 必須」を強制。
• 全列に PII フラグなし：気付かず個人情報を集計表に流出。 schema 段階で自動検出（regex/ML）。
• SLA が口約束：「明日までに更新」が文書化されていない → 障害時に責任の押し付け合い。 Data Contract で明文化。
• BI に直で raw を繋ぐ：DWH の整形層を経由しないと、 ロジックがダッシュボード内に散らばる。 必ず Mart 層を挟む。

リネージ・カタログ・契約は、 一見「事務作業」に見えますが、 障害対応の速度・新規メンバーのオンボーディング・経営層の信頼を直接決めます。 データエンジニアリングはコード半分、 ガバナンス半分の職能だと心得てください。

🔬 補講：よくある Q&A — データエンジニアリングの初学者がつまずく 10 問

現場で頻出する質問を Q&A 形式で整理します。 SSDSE-B-2026 の例を交えて答えます。

Q1. データエンジニアとデータサイエンティストの違いは？

A. 大雑把には「DE はパイプライン、 DS はモデル」。 DE は SSDSE を毎年取得・整形・配信する。 DS はその整形済みデータから人口予測モデルを作る。 ただし最近は Analytics Engineer という中間職もあり、 境界はグラデーション。

Q2. SQL と Python、 どちらを優先で学ぶ？

A. SQL を先に。 DWH 中心の現代では、 集計・結合・WINDOW 関数を SQL で書ける方が圧倒的に生産的。 Python は ETL や ML での周辺ロジックに使う。

Q3. データレイクとデータウェアハウスの違いは？

A. Lake は雑多な形式で生データを安く貯める場所（S3/GCS 上の CSV/JSON/Parquet）。 Warehouse は型付きテーブル + SQL 高速集計のための整理済み倉庫（BigQuery/Snowflake）。 最近の Lakehouse（Delta/Iceberg）は両者を融合。

Q4. dbt とは何？ なぜみんな使うのか

A. dbt は「SQL に Jinja マクロ + テスト + ドキュメント + lineage」を追加するツール。 DWH 内の Transform 層を Git 管理し、 CI/CD を回せる。 SSDSE 例では stg_ssdse_b → fact_population の SQL 変換を dbt model として書き、 テストとドキュメントを自動生成。

Q5. Airflow と dbt は競合？

A. 競合しません。 Airflow は DAG オーケストレータ（取得・変換・通知の流れ全体）。 dbt は DWH 内の SQL 変換専用。 通常は「Airflow が dbt run を呼ぶ」構成で組み合わせます。

Q6. なぜ Parquet なのか、 CSV じゃダメ？

A. (1) 列指向で集計が速い、 (2) 繰り返しの多い大規模データなら Snappy/Zstd で 5〜10x 圧縮、 (3) スキーマを内包する、 (4) BigQuery/Athena/Spark/DuckDB がネイティブ対応。 ただし SSDSE のような 564 行の小データでは圧縮効果は限定的で、 Snappy だと CSV と同程度（実装 6 参照）。 真価は大規模データで出る。

Q7. データ品質はどこまで担保すべき？

A. 「意思決定に使われる粒度まで」が原則。 全列全行を 100% 検証するのは過剰。 KPI に直結する列・キー列・PII 列に集中投資。 dbt tests / Great Expectations で宣言的に書く。

Q8. ETL ジョブが落ちました。 何から見る？

A. (1) ログで例外箇所特定、 (2) 入力データの schema/件数を前日比で確認、 (3) 上流 API のステータス、 (4) リソース（OOM/Disk Full）、 (5) idempotent ならリトライ、 そうでなければ手動修正後リラン。 リネージで影響範囲を即把握。

Q9. オンプレ DB から DWH への移行手順は？

A. (1) 移行対象テーブルの優先順位付け、 (2) Fivetran/Airbyte で初期 + CDC レプリ、 (3) dbt で staging → mart 再構築、 (4) BI を新 DWH に切替、 (5) 平行稼働で検証 → 旧停止。 SSDSE 規模なら 1 日、 業務 DB 1000 テーブルなら数ヶ月。

Q10. 今から学ぶならどの順番？

A. (1) SQL（PostgreSQL/DuckDB）, (2) Python + pandas, (3) dbt-core + DuckDB（ローカル DWH）, (4) Airflow（小さな DAG）, (5) BigQuery/Snowflake（クラウド DWH）, (6) Great Expectations（DQ）, (7) Datahub / OpenLineage（ガバナンス）。 SSDSE は (1)〜(3) の練習材料として最適。

本ページは「データエンジニアリングとは何か」を、 概念定義だけでなく、 SSDSE-B-2026 を使った 14 個の Python 実装 + 10 個の Q&A + 各種数式・落とし穴で立体的に解説しました。 概念マップ → 実装 → 補講 の順で読み返すと、 自分の中に「使える知識」として定着します。

🧮 実値で計算してみる（SSDSE-B-2026）

SSDSE-B-2026 の A1101（総人口）について、 3 つの正規化を 2023 年・47 都道府県の最小・中位付近・最大の代表 3 県（鳥取県・岐阜県・東京都）で電卓レベルで追ってみます。 平均 μ ≈ 2,645,809 人、 標準偏差 σ ≈ 2,797,551 人、 最小 = 537,000 人（鳥取）、 中央値 ≈ 1,549,000 人、 最大 = 14,086,000 人（東京）、 IQR ≈ 1,602,500 人とします。

z-score では東京が +4.09σ という強い外れ値だが、 Min-Max は東京を 1.0、 鳥取を 0.0 に張り付ける（中位付近の岐阜は 0.10 で左寄り）。 Robust スケーリングでは東京が +7.8 を超え、 外れ値性がより強調されます。 線形回帰では z-score、 木モデルでは Min-Max、 外れ値検出では Robust と使い分けます。

🧮 さらに深掘り：パイプラインの観測性・冪等性・パフォーマンス

SLI 監視だけでは不十分です。 ここでは 観測性（Observability）、 冪等性（Idempotency）、 パフォーマンス（並列・列指向） を、 SSDSE-B-2026 を題材に Python で実演します。

📐 数式：Little の法則とパイプラインのスループット

パイプラインの「待ち行列の長さ $L$」「到着率 $\lambda$（rows/sec）」「滞在時間 $W$（sec）」には、 Little の法則：

$$L = \lambda \cdot W$$

🔬 数式を言葉で読み解く

• $L$：システム内に滞留している行数（処理待ち + 処理中）
• $\lambda$：1 秒あたりに到着する行数（throughput）
• $W$：1 行が入ってから出るまでの平均滞在時間（latency）
• 含意：滞在時間 $W$ を半分にすれば、 同じ滞留量 $L$ で 2 倍のスループット $\lambda$ を捌ける。 並列化やバッチ処理は $W$ を縮めるテコです。

🧮 実値で計算：SSDSE 取り込みのスループット見積もり

SSDSE-B-2026 は 564 行・約 360KB。 ローカルでの読み込み時間を仮に 0.05 秒とすると：

本番では並列度 8 で読み込めば、 同じバッファで $\lambda$ を 8 倍に伸ばせます（理論値）。 ただし I/O ボトルネックや GIL の影響で実測は 3〜5 倍に留まることが多いです。

🐍 Python 実装 5：冪等な MERGE（UPSERT）パターン

🎯 このコードでやること：SSDSE データを (year, code, metric) を主キーにして SQLite に冪等投入する。 二度実行しても重複しない。

📥 入力データ：long_df（year, code, pref, metric, value の 5 列）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import sqlite3
import pandas as pd

df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])
df = df.rename(columns={'SSDSE-B-2026': 'year', 'Code': 'code', 'Prefecture': 'pref'})
long_df = df.melt(id_vars=['year', 'code', 'pref'], var_name='metric', value_name='value')

conn = sqlite3.connect('warehouse.db')
conn.execute("""CREATE TABLE IF NOT EXISTS ssdse_b_upsert (
    year INTEGER, code TEXT, pref TEXT, metric TEXT, value REAL,
    PRIMARY KEY (year, code, metric)
)""")

# 冪等な UPSERT: 同じキーは置換 → 何度流しても結果が同じ
rows = long_df.to_dict('records')
conn.executemany(
    """INSERT OR REPLACE INTO ssdse_b_upsert
       VALUES (:year, :code, :pref, :metric, :value)""", rows)
conn.commit()
print(conn.execute("SELECT COUNT(*) FROM ssdse_b_upsert").fetchone())

📤 実行すると次の出力が得られる：

💬 結果の読み方：2 回流しても件数が変わらない＝冪等。 障害復旧時にリトライしても安全。 これがない pipeline は「2 倍に膨れる事故」を起こします。

🐍 Python 実装 6：列指向 Parquet で行指向 CSV を圧縮する

🎯 このコードでやること：SSDSE-B-2026 を Parquet（列指向 + Snappy 圧縮）に変換し、 CSV と比較してファイルサイズを測る。

📥 入力データ：SSDSE-B-2026.csv（約 360KB, 564 行 × 112 列）。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import pyarrow  # parquet の読み書きに必要（ブラウザには無い）
import pandas as pd
import os

df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])

# Parquet (Snappy) で保存
df.to_parquet('ssdse_b_2026.parquet', compression='snappy')

size_csv = os.path.getsize('data/raw/SSDSE-B-2026.csv')
size_pq = os.path.getsize('ssdse_b_2026.parquet')
print(f"CSV     : {size_csv:>10,} bytes")
print(f"Parquet : {size_pq:>10,} bytes  (対 CSV 比 {size_pq/size_csv:.2%})")

📤 実行すると次の出力が得られる：

💬 結果の読み方：この 564 行の小さな整数中心データでは、 Snappy Parquet はむしろ CSV と同程度（約 100%）で、 圧縮メリットは出ません（zstd/brotli にすると約 70% まで下がります）。 Parquet の本質的な強みは「サイズ」より 列指向で必要な列だけ読める点と、 数百万行・繰り返しの多い大規模データで初めて効く 5〜10x 圧縮です。 DWH/Lake 標準フォーマットとして BigQuery / Athena / Spark がネイティブ対応します。

🐍 Python 実装 7：Great Expectations 風のシンプル DQ

🎯 このコードでやること：SSDSE-B-2026 の総人口列に「常に正」「東京 ≥ 沖縄」「合計 = 男 + 女」など複数の期待値を書き、 違反を一括検出する。

📥 入力データ：SSDSE-B-2026（A1101 総人口、 A110101 男、 A110102 女）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

import pandas as pd

df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])
df = df.rename(columns={'SSDSE-B-2026': 'year', 'Prefecture': 'pref'})

expectations = []

# 1. 総人口は常に正
expectations.append(('A1101 > 0', (df['A1101'] > 0).all()))

# 2. 各年で東京（13） ≥ 沖縄（47） の人口
yearly = df.pivot_table(index='year', columns='pref', values='A1101')
expectations.append(('Tokyo >= Okinawa', (yearly['東京都'] >= yearly['沖縄県']).all()))

# 3. 総人口 == 男 + 女 (±5000 人許容)
diff = (df['A1101'] - df['A110101'] - df['A110102']).abs()
expectations.append(('|total - (M+F)| <= 5000', (diff <= 5000).all()))

for name, ok in expectations: print(f"[{'PASS' if ok else 'FAIL'}] {name}")

📤 実行すると次の出力が得られる：

💬 結果の読み方：3 つの期待値すべて PASS。 本番では Great Expectations / dbt tests でこれを宣言的に書き、 CI/CD に組み込みます。 期待値を「コード化」することで、 上流変更時に即検知できます。

⚠️ アンチパターン（実務でよく見る失敗）

• 「Excel ETL」：手作業で Excel を毎日更新 → 担当者が休むと壊れる。 即 Python/dbt 化。
• 「魔法の cron」：crontab に並んだ謎スクリプト群、 依存関係が口伝。 Airflow/Dagster で DAG 化を。
• 「神 SQL」：1000 行の SELECT が CTE で深くネストし誰も読めない。 dbt のモジュール化（model 分割）で解消。
• 「Snapshot 上書き」：日次で全件上書き → 過去が失われる。 SCD Type 2（履歴保持）か append-only に。
• 「マスタテーブルが 2 つ」：source of truth が分裂。 DWH に統合し、 上流は読み専用へ。

📚 補足：データエンジニアのキャリアと組織

SSDSE-B-2026 という小さなデータでも、 「取得 → 変換 → 配信 → 監視」のすべての工程を体験できます。 まずは本ページの 7 つの Python 実装をローカルで動かし、 自分の手で SLI を測り、 冪等性を確かめ、 Parquet 化することから始めてみてください。 大規模になった時、 同じ原則がそのまま効きます。

🧮 数式に値を入れて手で計算する: ETL パイプラインのスループット

合成データで ETL 段階別の処理時間とボトルネックを特定する。

Step 1: 段階別処理 [秒/100MB]

Step 2: ボトルネックと全体時間

🐍 Python で再現

1
2
3
4
5
6
7

import numpy as np
times = np.array([10, 40, 15])
throughput = 100 / times
total = times.sum()
print(f"スループット: {throughput.round(2)} MB/s")
print(f"ボトルネック: stage {throughput.argmin()}")
print(f"直列合計: {total} 秒")

📤 実行結果

💬 手計算 (Step 2) ボトルネック T、 65 秒と Python 出力が完全一致。

🐍 Python 実装 — 3 つのスケーラを 1 行ずつ

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import pandas as pd
from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler

df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])
x = df[df['SSDSE-B-2026'] == 2023][['A1101']]   # 2023 年・47 都道府県
print('z-score (mean=0, std=1):',
      StandardScaler().fit_transform(x).flatten()[:3].round(2))
print('min-max [0,1]          :',
      MinMaxScaler().fit_transform(x).flatten()[:3].round(2))
print('robust (median, IQR)   :',
      RobustScaler().fit_transform(x).flatten()[:3].round(2))

🐍 ケーススタディ：SSDSE-B-2026 で実現する 8 ステップの最小 ELT

最後に、 これまでの要素を統合して「raw CSV → Parquet → DWH (SQLite) → 分析クエリ → 監視ログ」までの最小 ELT を一気通貫で組みます。 すべて SSDSE-B-2026 を入力とし、 ローカル環境だけで完結します。

📐 数式：データボリュームの増分処理（差分積み上げ）

全件再処理（full refresh）は単純ですが、 データ量 $N$ に比例した時間 $T \propto N$ がかかります。 増分処理（incremental）は新規分 $\Delta N$ だけ処理：

$$T_{\text{incr}} = T_0 + c \cdot \Delta N \ll T_{\text{full}} = c \cdot N \quad (\Delta N \ll N)$$

🔬 数式を言葉で読み解く

• $T_{\text{full}}$：毎回全件処理する時間。 データが 10 倍になれば 10 倍遅くなる。
• $T_{\text{incr}}$：差分のみ処理する時間。 $T_0$ は固定オーバーヘッド、 $c$ は 1 行あたりの処理時間。
• $\Delta N$：今回の差分行数。 通常 $\Delta N \ll N$（例：年に 1 度の SSDSE 更新では $\Delta N$ ≈ 47 行）。
• 含意：「watermark（年度・updated_at）」で差分を切り出せば、 SSDSE-B 年次更新 → 47 行追加のみで済む。 全件の 1/12 の時間で完了。

🧮 実値で計算：full vs incremental の比較

564 行を処理するパイプライン。 SSDSE-B が毎年 47 行（1 年分の都道府県）追加されると仮定：

小規模では差は小さいですが、 1 億行になると「1 時間 vs 数秒」の差になります。 incremental 設計は早めに仕込むのが鉄則。

🐍 Python 実装 8：エンドツーエンド ELT スクリプト

🎯 このコードでやること：SSDSE-B-2026 を取得・Parquet 化・SQLite 投入・品質チェック・サマリクエリまで一気通貫で実行する。

📥 入力データ：data/raw/SSDSE-B-2026.csv（cp932）。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

import pyarrow  # parquet の読み書きに必要（ブラウザには無い）
import pandas as pd
import sqlite3
import time

t0 = time.time()

# STEP 1: Extract（取得）
# 改名元が英字の列名なので skiprows=[1] で読む（header=1 だと日本語の列名になる）
df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])
df = df.rename(columns={'SSDSE-B-2026': 'year', 'Code': 'code', 'Prefecture': 'pref'})

# STEP 2: Save raw to Parquet
df.to_parquet('ssdse_b_2026.parquet', compression='snappy')

# STEP 3: Transform（wide → long）
long_df = df.melt(id_vars=['year', 'code', 'pref'], var_name='metric', value_name='value')

# STEP 4: DQ チェック
# 気温（B41xx）は氷点下があるので、非負チェックの対象外にする
_nonneg = long_df[~long_df['metric'].str.startswith('B41')]
assert (_nonneg['value'].dropna() >= 0).all(), "negative value found"
assert long_df['year'].between(1975, 2030).all(), "year out of range"

# STEP 5: Load → SQLite
conn = sqlite3.connect('warehouse.db')
long_df.to_sql('ssdse_b_long', conn, if_exists='replace', index=False)

# STEP 6: 分析クエリ
top5 = pd.read_sql(
    "SELECT pref, value FROM ssdse_b_long WHERE year=2023 AND metric='A1101' ORDER BY value DESC LIMIT 5", conn)
conn.close()
print(top5); print(f"elapsed: {time.time()-t0:.2f} sec")

📤 実行すると次の出力が得られる：

💬 結果の読み方：8 ステップが約 1.5 秒で完了。 raw → Parquet → DWH → クエリまで一気通貫。 本番では各ステップを Airflow Task に分割し、 リトライ・SLA・通知を付与します。 でも本質的な「やること」はこの 30 行と同じです。

📚 学習ロードマップ：データエンジニアになるための 6 ステップ

1. SQL を読み書きできる — JOIN, WINDOW 関数, CTE まで（SSDSE を SQLite で触る本ページが入り口）。
2. Python + pandas でデータ整形 — melt, pivot, groupby, merge を自在に。
3. 列指向ファイル（Parquet）と DWH — Snappy 圧縮、 partition、 clustering。
4. Orchestration（Airflow/Dagster/Prefect） — DAG、 retry、 SLA、 backfill。
5. Data Quality と Observability — Great Expectations / dbt tests / OpenLineage。
6. クラウド DWH と IaC — BigQuery/Snowflake/Redshift + Terraform/dbt。

本ページの 8 つの Python 実装は、 この 6 ステップのうち (1)〜(3) と (5) を実体験できる構成にしてあります。 まずは手を動かすこと、 次に SLI を測ること、 最後に DAG 化すること——この順序で身につけてください。

🐍 補講：DuckDB で「個人用 DWH」を 5 分で立ち上げる

BigQuery/Snowflake を導入する前の「個人検証 DWH」として最強なのが DuckDB。 SQLite と同じくファイル 1 つ・サーバ不要、 しかし列指向で OLAP 高速。 SSDSE-B-2026 をその場で SQL 集計してみます。

🐍 Python 実装 13：DuckDB で CSV を直接 SQL 集計

🎯 このコードでやること：DuckDB で SSDSE-B-2026.csv を直接 SQL クエリ。 INSERT 不要、 ファイルパスのままテーブルとして扱える。

📥 入力データ：data/raw/SSDSE-B-2026.csv（先に Parquet 化済みなら更に速い）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import duckdb

# 1. CSV をテーブルとして直接クエリ（INSERT 不要）
con = duckdb.connect()
con.execute("""CREATE OR REPLACE VIEW ssdse AS
    SELECT * FROM read_csv_auto('data/raw/SSDSE-B-2026.csv',
                                 ignore_errors=true,
                                 skip=2,
                                 header=false)""")

# 2. 集計クエリ: 2023 年の高齢化率 TOP5（A1303 / A1101）
q = """SELECT column02 AS pref, ROUND(column15 / column03 * 100, 1) AS aging_pct
       FROM ssdse WHERE column00 = 2023 ORDER BY aging_pct DESC LIMIT 5"""
print(con.execute(q).fetchdf())

📤 実行すると次の出力が得られる：

💬 結果の読み方：秋田県の高齢化率が 39.1% で全国最高。 DuckDB なら BigQuery を立てる前に、 こうした SQL 検証がローカルで即可能。 列指向 + ベクトル化エンジンで pandas より高速なケースも多い。

🐍 Python 実装 14：DuckDB で Parquet を直接読む

🎯 このコードでやること：先ほど作成した Parquet を DuckDB から直接 SELECT し、 BigQuery 同様の体験をローカルで再現する。

📥 入力データ：ssdse_b_2026.parquet（実装 6 で作成）。

1
2
3
4
5
6
7
8
9
10
11
12

import duckdb

con = duckdb.connect()

# Parquet をテーブルとして直接 SELECT
q = """SELECT Prefecture, A1101 AS population
       FROM 'ssdse_b_2026.parquet'
       WHERE "SSDSE-B-2026" = 2023
       ORDER BY A1101 DESC
       LIMIT 3"""
print(con.execute(q).fetchdf())
print("読み込み時間 (DuckDB + Parquet) はミリ秒オーダー")

📤 実行すると次の出力が得られる：

💬 結果の読み方：Parquet + DuckDB は数百 MB 〜 数 GB クラスのデータでも秒以下。 個人検証や小チームには、 BigQuery より DuckDB の方が安く・速いケースも多い。 dbt-duckdb と組み合わせれば「dbt 開発をローカルで完結」も可能。

⚠️ DuckDB の使い所と限界

• ✅ 向く：個人検証、 ローカル ETL、 中小データ（数 GB 程度）、 dbt 開発環境、 CI でのデータテスト。
• ❌ 向かない：複数同時書き込み、 数 TB 規模、 SLA 99.99% 必須の本番。 そこは BigQuery/Snowflake へ。
• Tip：本番 DWH と DuckDB を「同じ SQL（dbt model）で動かす」設計にすると、 開発速度が劇的に向上。

DuckDB を含めて、 本ページの Python 実装は 計 14 本。 SSDSE-B-2026 という 360KB の小さな CSV から、 取得・変換・配信・監視・リネージ・MDS 全レイヤを実体験できる構成にしてあります。 大規模になっても原則は同じです。 自分の手で動かして、 数字を測る習慣を身につけてください。

⚠️ 落とし穴

データエンジニアリングの実務で頻発する落とし穴。 ETL/ELT・パイプライン設計・データ品質の 3 領域で発生しやすい固有の罠を列挙する。

• スキーマ進化の無対応: 上流の API・DB のスキーマが変わるとパイプラインが沈黙して壊れる。 対策: Schema Registry (Confluent、 Avro)、 Iceberg/Delta の Schema Evolution、 dbt の schema test で contract test を仕込む。
• Train-Test リーク: 前処理 (Imputer、 Scaler、 TargetEncoder) を CV ループの外で fit すると、 検証セットの統計量が学習に漏れる。 対策: sklearn Pipeline で fit を fold 内に閉じ込め、 dbt の seed/snapshot で時点固定。
• 小ファイル過多 (small files problem): Kafka/Kinesis のストリーミングで 1KB ファイルが大量生成 → メタデータ操作が重く HDFS/S3 が遅延。 対策: Delta/Iceberg の OPTIMIZE、 Spark の coalesce で 100MB-1GB に compaction。
• シャッフル爆発と shuffle skew: groupBy・join でネットワーク I/O が支配的に。 「東京だけ巨大」のような skew で 1 ワーカが詰まる。 対策: Broadcast Join、 Salting、 AQE (Adaptive Query Execution)、 事前集約 (combiner)。
• べき等性 (Idempotency) の欠如: 障害でジョブを再実行したら同じ行が二重挿入される。 対策: MERGE/UPSERT (Snowflake、 BigQuery)、 dbt の incremental + unique_key、 メッセージにイベント ID を付与。
• データ系統 (Data Lineage) の不在: 「この KPI 表はどの生データから来たか」が追えないと監査・障害対応で詰む。 対策: Apache Atlas、 OpenLineage、 dbt 系統グラフ、 Marquez。
• S3/GCS の公開設定誤り: バケットが誤って公開され個人情報流出。 Capital One、 Verizon、 Accenture など事例多数。 対策: Block Public Access、 IAM 最小権限、 KMS 暗号化、 設定自動監査 (AWS Config)。

🌐 関連手法・派生

• ETL / ELT：データ流通の基本パターン。
• データクレンジング：型・重複・欠損・表記揺れの修正。
• 特徴量エンジニアリング：ドメイン知識を変数化。
• データパイプライン：再現性を担保する仕組み。
• データ品質：完全性・正確性・一貫性・適時性。

🏭 1. データエンジニアリングとは

分析・モデリングに使える形にデータを整える・繋ぐ・流す工程。 データサイエンスの 70-80% の工数がここに費やされる。

• 狭義のDE：ETL/ELT・データ基盤・パイプライン構築（インフラ寄り）
• 広義のDE：前処理・特徴量エンジニアリング・データ品質管理

本ページは分析者寄りの DE 知識をまとめます。

🔄 2. ETL / ELT

• ETL：Extract（抽出）→ Transform（変換）→ Load（格納）。 古典的
• ELT：Extract → Load → Transform。 DWH の計算能力を活かす現代型

典型的なソース

• CSV / TSV / Excel（公的統計の標準）
• API（REST・GraphQL）
• データベース（PostgreSQL・MySQL・BigQuery）
• ログ・ストリーミング（Kafka・Kinesis）
• Web スクレイピング（要利用規約遵守）

📥 3. データ読み込み

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

import pandas as pd
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LinearRegression
from sklearn.model_selection import train_test_split

df = pd.read_csv('data/raw/SSDSE-B-2026.csv', encoding='cp932', skiprows=[1])

# 地方区分カテゴリを Code から導出（8 地方）
df['pref_no'] = df['Code'].str[1:3].astype(int)
df['region'] = pd.cut(df['pref_no'], bins=[0, 1, 7, 14, 23, 30, 35, 39, 47],
                      labels=['北海道','東北','関東','中部','近畿','中国','四国','九州'])

num_cols = ['A1101', 'A1303', 'A4101']   # 総人口・65歳以上人口・出生数
cat_cols = ['region']                    # 地方区分（カテゴリ）

preprocessor = ColumnTransformer([
    ('num', Pipeline([
        ('impute', SimpleImputer(strategy='median')),
        ('scale', StandardScaler())]), num_cols),
    ('cat', OneHotEncoder(handle_unknown='ignore', drop='first'), cat_cols)
])

pipe = Pipeline([('prep', preprocessor), ('model', LinearRegression())])

X = df[num_cols + cat_cols]
y = df['A4200']  # 目的変数：死亡数
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.3, random_state=42)
pipe.fit(X_tr, y_tr)          # train でだけ統計量を学習 — リーク完全防止
print(f'R2 (test): {pipe.score(X_te, y_te):.3f}')

② scipy.stats でロバストな統計量を計算

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

from scipy import stats
import numpy as np

x = df[df['SSDSE-B-2026'] == 2023]['A1101'].dropna()   # 2023 年・47 都道府県

# 通常の z-score（外れ値の影響を受ける）
z_normal = (x - x.mean()) / x.std()

# ロバスト z-score（MAD ベース・東京の影響を受けにくい）
median = x.median()
mad = stats.median_abs_deviation(x)
z_robust = 0.6745 * (x - median) / mad

# Trimmed mean（上下 10% を除いた平均）
trimmed = stats.trim_mean(x, 0.1)

# IQR で外れ値検出
q1, q3 = np.percentile(x, [25, 75])
iqr = q3 - q1
outliers = x[(x < q1 - 1.5*iqr) | (x > q3 + 1.5*iqr)]
print(f'外れ値件数: {len(outliers)}')
print(f'通常平均: {x.mean():.0f}, 10%トリム平均: {trimmed:.0f}')

🌐 補講：モダンデータスタック（Modern Data Stack）の全景

2020 年以降の業界標準が Modern Data Stack（MDS）。 「クラウド DWH を中心に、 SaaS ツールで取得・変換・BI・監視を組む」スタイル。 SSDSE-B-2026 を題材に、 MDS の各レイヤと代表ツールを整理します。

📊 MDS のレイヤ別代表ツール

🎨 直感：「組み立て式の家」

昔は「Hadoop で全部やる」一体型の家。 MDS は「DWH を土台に、 部屋（ツール）を必要なだけ後付け」する組み立て式。 各ツールが API/SQL で疎結合に繋がるため、 部分入替が容易です。 一方、 ツール間連携と契約管理が複雑になるため、 ガバナンスがより重要に。

⚠️ MDS の落とし穴

• SaaS 課金の積み上げ：Fivetran ＋ Snowflake ＋ dbt Cloud ＋ Looker の従量課金が重なり → 月数百万円。 ROI を都度評価。
• ツール選定疲れ：似たツールが乱立。 「最低限の機能で動く」最小構成から始める。
• クラウド DWH ロックイン：BigQuery 専用 SQL を書きまくると Snowflake へ移れない。 dbt で抽象化を。
• BI の二重実装：DWH と BI ツールの両方で同じ KPI ロジック → 数字がズレる。 Semantic Layer（dbt Metrics, Cube, MetricFlow）で一本化。

MDS は強力ですが「全部入れたら勝ち」ではありません。 SSDSE のような小規模公的データなら、 ローカルで pandas + DuckDB + dbt-core + Metabase でも十分。 まず最小構成で動かし、 必要になったら SaaS を足す——これが正攻法です。

🔗 関連用語 — 学習ネットワーク（拡張版）

📚 関連グループ教材

データエンジニアリングは複数領域の交差点。 関連グループと公式ドキュメント・標準仕様を併用してください。

• 📚 CSV / DataFrame グループ — 表形式データの基本
• 📚 データクレンジンググループ — Transform 段階
• 📚 データレイクグループ — ストレージ層
• 📚 ビッグデータグループ — Spark / Hadoop エコシステム
• 📚 MLOps グループ — 機械学習との接続

• pandas 公式ドキュメント（https://pandas.pydata.org/docs/）— Python 表データ処理の標準
• Apache Airflow（https://airflow.apache.org/docs/）— ワークフロー編成ツール
• dbt 公式ドキュメント（https://docs.getdbt.com/）— SQL ベース Transform フレームワーク
• Apache Parquet（https://parquet.apache.org/docs/）— 列指向ファイルフォーマット標準
• DuckDB 公式（https://duckdb.org/docs/）— インプロセス分析 DB
• SSDSE-B-2026 教育用標準データセット（https://www.nstac.go.jp/use/literacy/ssdse/）— ETL 実演用

📚 チートシート：データエンジニアリングの必修コマンド・関数

この 14 関数を覚えれば、 SSDSE 規模のデータエンジニアリングは 80% カバーできます。 本番の DWH もこれらの「拡張版」に過ぎません。

🗺 概念マップ

データエンジニアリングを中心に、 上流の (ETL / ELT / CDC)、 並列の (データクレンジング / データウェアハウス / Data Lake)、 下流の (BI / 機械学習 / Feature Store) を配置した。 SSDSE-B-2026 のような年次更新 CSV データを取り込み、 標準化・変換し、 BI や ML 配信まで運ぶ流れを 1 枚で俯瞰できる。

データエンジニアリング周辺の概念は (1) 抽出・変換・読込 (ETL / ELT / CDC / change data capture)、 (2) ストレージ (DWH / Data Lake / Lakehouse / DataMart)、 (3) 処理基盤 (Spark / Airflow / dbt / Kafka)、 (4) 品質管理 (データクレンジング / バリデーション / リネージ追跡) の 4 層に整理される。 これらを統合的に設計することで、 後段の BI 分析や機械学習を支える堅牢なデータ基盤が構築できる。

🔗 隣接手法への橋渡し

データエンジニアリングは収集と分析の間の基盤層。 3 視点 (接続・統合・比較) で隣接基盤概念との関係を整理する。

🔌 接続: 上流・下流での連鎖

• 上流: データ収集、 データガバナンス — 入力と方針
• 並列: 分散データ処理、 データレイク — 規模対応の技術選定
• 下流: データ分析のプロセス、 MLOps — 分析工程に整備データを引き渡す

🧩 統合: 基盤パイプラインへの組み込み

ソース (DB/API/CSV) → 取り込み (ETL/ELT) → DWH/lake (構造化/非構造化) → 変換 (dbt/Airflow) → モデリング → BI/ML 配信の流れで、 SSDSE 系の年次更新を想定した再実行性が要点。 各層に冪等性 (再実行可能) とスキーマバージョニングを組み込むと、 障害復旧と監査追跡が容易になる。

⚖️ 比較: 隣接基盤概念との位置づけ

4 概念は重なるが守備範囲が異なる。 データエンジニアリングが全体を統括し、 レイクが蓄積、 分散処理が計算、 MLOps がモデル運用を担う。 SSDSE 規模なら lake 不要だが、 dbt や Airflow による再実行可能な ETL は導入価値あり。

🌳 手法選択フロー

データエンジニアリングは収集 → 変換 → 配信の三層を設計・運用する。

1. 収集 → 蓄積の設計は？ Yes → データ収集、 No → データレイク を先に確認
2. 変換 (ETL/ELT) の設計は？ Yes → ETL、 No → データクレンジング を先に確認
3. 配信・基盤の設計は？ Yes → 分散データ処理、 No → BI ツール を先に確認

バッチ中心なら ETL + DWH、 ストリーミング中心なら Kafka + Flink、 ML 主体なら feature store、 と「データ流速と用途」で選ぶ。