Parquetファイルをデータパイプラインの中間形式に選んだ理由

1
import pandas as pd
2

3
# CSVに書き出し
4
df = pd.DataFrame({"value": [145.50, 146.20], "count": [1000, 2000]})
5
df.to_csv("data.csv", index=False)
6

7
# CSVから読み込み → 型が変わることがある
8
df_loaded = pd.read_csv("data.csv")
9
print(df_loaded.dtypes)
10
# value     float64  ← たまたま正しい
11
# count       int64  ← たまたま正しい
12

13
# しかし、欠損値があると...
14
df2 = pd.DataFrame({"value": [145.50, None], "count": [1000, None]})
15
df2.to_csv("data2.csv", index=False)
4 collapsed lines
16
df2_loaded = pd.read_csv("data2.csv")
17
print(df2_loaded.dtypes)
18
# value     float64
19
# count     float64  ← intがfloatに変わった

1
# CSV: 411カラム全部読んでからフィルタ
2
df = pd.read_csv("analysis.csv")  # 全カラムパース（遅い）
3
df = df[["timestamp", "metric_a", "metric_b"]]  # 使うのは3カラムだけ

1
import pandas as pd
2

3
# Parquet: 必要な列だけ読み込み（高速）
4
df = pd.read_parquet(
5
    "analysis_results.parquet",
6
    columns=["timestamp", "metric_a", "metric_b"]
7
)

1
import pyarrow.parquet as pq
2

3
# 書き込み（型情報付き）
4
df.to_parquet("analysis_results.parquet", engine="pyarrow")
5

6
# スキーマ確認
7
schema = pq.read_schema("analysis_results.parquet")
8
print(schema)
9
# timestamp: double
10
# metric_a: double
11
# metric_b: double
12
# result_label: string
13
# ... 型情報が保存されている

1
import os
2

3
# 同じデータをCSVとParquetで保存して比較
4
df.to_csv("data.csv", index=False)
5
df.to_parquet("data.parquet", engine="pyarrow", compression="snappy")
6

7
csv_size = os.path.getsize("data.csv")
8
parquet_size = os.path.getsize("data.parquet")
9
print(f"CSV: {csv_size / 1024 / 1024:.1f} MB")
10
print(f"Parquet: {parquet_size / 1024 / 1024:.1f} MB")
11
print(f"圧縮率: {parquet_size / csv_size:.1%}")
12
# 実測: CSV 230MB → Parquet 28MB（約12%）

1
# Before
2
df.to_csv("output.csv", index=False)
3
df = pd.read_csv("output.csv")
4

5
# After
6
df.to_parquet("output.parquet", engine="pyarrow")
7
df = pd.read_parquet("output.parquet")

1
411カラム（分析指標）
2
  x 10データソース
3
  x 8集計期間（1H, 2H, 4H, 6H, 8H, 12H, D, W）
4
  = 約32,880カラム相当のデータ量

1
from datetime import datetime
2

3

4
def add_reference_columns(df: pd.DataFrame, source: str, period: str) -> pd.DataFrame:
5
    """BIツール連携用のメタデータカラムを追加する。"""
6
    df = df.copy()
7
    df["ReferenceDate"] = datetime.now().strftime("%Y-%m-%d")
8
    df["Source"] = source
9
    df["Period"] = period
10
    return df

1
from datetime import datetime
2

3

4
def generate_composite_id(
5
    reference_date: str,
6
    source: str,
7
    period: str,
8
    sequence: int,
9
) -> str:
10
    """複合IDを生成する。
11

12
    形式: {YYYYMMDD}_{Source}_{Period}_{YYYYMMDDHHMM}_{00001-99999}
13
    例: 20260327_SRC001_4H_202603271500_00001
14
    """
15
    now = datetime.now().strftime("%Y%m%d%H%M")
1 collapsed line
16
    return f"{reference_date}_{source}_{period}_{now}_{sequence:05d}"

1
Before: analysis_SRC001_4H_20260301.csv
2
After:  analysis_results_SRC001_4H_20260301.parquet

1
## IMPACT_ANALYSIS 必須チェック項目（データ形式変更時）
2

3
1. このファイルを読み込んでいるスクリプトの一覧
4
2. ファイル名パターンをハードコードしている箇所
5
3. 拡張子でフィルタしている箇所（*.csv → *.parquet）
6
4. バックアップスクリプトの対象パターン
7
5. CI/CDパイプラインの成果物パス