バラバラな売上データを正規化して統合するETLパイプライン（Python × SQL）

## 🎯 背景
複数の店舗から届く売上データのフォーマットがバラバラで、
集計や分析の前に毎回「整形・確認・結合」といった前処理に時間がかかる――
そんな状況を想定して、実務で役立つETLパイプラインの構築に取り組みました。


支店や店舗単位で売上データが提出されるような業務（例：小売・飲食・フランチャイズ運営など）では、 形式の異なるCSVが毎月集まり、整形・集計に毎回工数がかかるケースがあると思います。

本ツールは、そうした 「バラバラな売上データを扱う実務現場」を想定して設計し、異なるCSVを統一→DBに格納→可視化まで一気通貫で行えるETLパイプラインを構築しました。

データエンジニアを志す中で、こうした「分析可能な形にデータを整える力」を強化したくて取り組んだプロジェクトです。

## ⏱ Before / After（改善効果）

| 項目 | Before（従来の業務） | After（本ツール使用後） |
|------|----------------------|--------------------------|
| データ整形 | フォーマット確認・手作業で統一 | `normalize.py`で自動整形（5分未満） |
| 集計処理 | Excelで関数・ピボット | DB＋SQLで自動抽出 |
| 可視化・共有 | スクショやExcelグラフ貼付 | Streamlitで即グラフ表示・共有可能 |
| 総作業時間 | 約1.5〜2時間 | 約10分以内に短縮  |

## 🛠 使用技術

- Python（pandas）  
- SQLite  
- Streamlit  
- Altair  
- Mermaid

今回は **軽量なSQLite + Streamlit** を選定しています。  
理由は、ローカルでもすぐ動かせるシンプルな構成であり、非エンジニアでも扱いやすく、「PoCや業務支援ツール」として実運用に近い形を模擬できるからです。
また、整形処理には **pandas** を活用し、**将来的なクラウドDB連携（BigQuery等）も見据えた設計**としています。



## 📁 構成図

### ETLパイプライン全体の流れ

```mermaid
graph TD
    A[CSVファイル] --> B[normalize.py（整形処理）]
    B --> C[normalized_sales.csv]
    C --> D[etl.pyでDB格納]
    D --> E[sales.db（3テーブル）]
    E --> F[Streamlitで読み込み]
    F --> G[Webで表示・分析]
```


### Streamlitアプリ内部の処理構造

```mermaid
graph TD
    A[SQLite接続] --> B[売上データ読込（pandas）]
    B --> C[店舗選択 selectbox]
    C --> D[選択店舗のフィルタリング]
    D --> E[DataFrame表示]
    D --> F[売上金額列の追加]
    F --> G[商品別にgroupby]
    G --> H[Altairでグラフ化]
    H --> I[注目商品を強調表示]
```

## 📦 課題と解決策
| 課題 | 解決方法 |
|------|----------|
| 列名の違い（Date vs 販売日） | normalize.pyで統一 |
| 商品名の表記ゆれ（Tシャツ, tshirt, シャツ） | 辞書で正規化 |
| 日付や数値の形式不統一 | pandas.to_datetimeで統一 |
| DB設計が曖昧 | schema.sqlで明示的に設計 |


## 📁 ディレクトリ構成
```text
ec_sales_pipeline/
├── data/
│   ├── store_a_2025.csv
│   ├── store_b_2025.csv
├── scripts/
│   ├── etl.py
│   ├── normalize.py
├── db/
│   ├── schema.sql
│   └── sales.db
├── output/
│   └── reports/
├── README.md
├── requirements.txt
└── .gitignore
```
## 🧹データ整形：normalize.pyの例（列名統一・商品名正規化）

```python
def normalize_file(filepath, store_name):
    df = pd.read_csv(filepath, encoding='utf-8-sig', quotechar='"')
    df.columns = [col.strip().lower() for col in df.columns]

    col_map = {
        'item': 'item_name', '商品名': 'item_name',
        'date': 'sale_date', '売上日': 'sale_date',
        'qty': 'quantity', '数量': 'quantity',
        'price': 'unit_price', '単価': 'unit_price'
    }
    df = df.rename(columns=col_map)

    df['item_name'] = df['item_name'].replace({
        'tshirt': 'Tシャツ', 'シャツ': 'Tシャツ'
    })

    df['sale_date'] = pd.to_datetime(df['sale_date'], errors='coerce').dt.strftime('%Y-%m-%d')
    df['store_name'] = store_name
    return df[['sale_date', 'item_name', 'quantity', 'unit_price', 'store_name']]

```
🏗️ DB設計：schema.sql
```
CREATE TABLE stores (
  id INTEGER PRIMARY KEY,
  name TEXT
);

CREATE TABLE products (
  id INTEGER PRIMARY KEY,
  name TEXT
);

CREATE TABLE sales (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  sale_date TEXT,
  store_id INTEGER,
  product_id INTEGER,
  quantity INTEGER,
  unit_price INTEGER,
  FOREIGN KEY (store_id) REFERENCES stores(id),
  FOREIGN KEY (product_id) REFERENCES products(id)
);
```
🔍 分析例：月別売上推移
```
SELECT
  strftime('%Y-%m', sale_date) AS month,
  p.name AS product,
  SUM(s.quantity * s.unit_price) AS total_sales
FROM sales s
JOIN products p ON s.product_id = p.id
GROUP BY month, product
ORDER BY month;
```
## 📺 Streamlitで売上データを可視化
📷 アプリ画面（Altairグラフ） → 商品別の売上金額をランキング表示。Shortsの売上が突出していたことが視覚的にわかります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/4024841/f4a1c1dc-fce3-4c70-a10e-ccd51bfaeb98.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/4024841/12b64d1a-f5e6-4dd9-b01d-ac58f6ef467c.png)
https://animated-robot-x57764xpqq5h9pj9-8502.app.github.dev/
グラフから「Shorts」の売上が突出していたことが分かりました。


🧪 Altairコード例
```
import pandas as pd
import sqlite3
import altair as alt
import streamlit as st

st.title("📊 売上データビューア")

conn = sqlite3.connect("db/sales.db")
query = """
SELECT sale_date, item_name, quantity, unit_price, store_name
FROM sales_raw
"""
df = pd.read_sql_query(query, conn)

df["売上"] = df["quantity"] * df["unit_price"]
st.dataframe(df)

summary = df.groupby("item_name", as_index=False)["売上"].sum()
chart = alt.Chart(summary).mark_bar().encode(
    x=alt.X("売上:Q", title="売上金額（円）"),
    y=alt.Y("item_name:N", sort='-x', title="商品名"),
    color=alt.condition(
        alt.datum.item_name == "Shorts", alt.value("orange"), alt.value("steelblue")
    )
).properties(title="商品別売上金額（Altairグラフ）")

st.altair_chart(chart, use_container_width=True)
```
## 🧵 学び・工夫・今後の展望

- **GitHub Actions** を活用して、定期的にデータ整形・更新処理を自動実行したい（例：週1でCSV更新＆Web再生成）
- SQLiteの代わりに **PostgreSQLやBigQuery** に対応することで、クラウド連携・データ量拡張にも対応
- アプリの汎用性を高めるため、複数商品カテゴリへの対応、売上比率の円グラフ化なども計画中


今後は GitHub Actions による定期実行や BigQuery対応などを通じて、  
「業務の中で本当に使えるデータ整備」の土台づくりを強化していきたいです。


🔗 GitHubリポジトリ
👉 https://github.com/flopsy212/Sales-etl-streamlit