概要（CSV のクリーニングは「人力のチェックを Python に任せる」作業）
CSV のクリーニングは、ざっくり言うと
- 余計な空白や文字の揺れを揃える
- 数値・日付をちゃんと数値・日付として扱えるように直す
- 欠損・重複・異常値を見つけて整理する
という「キタナイ生データ」を「機械が扱いやすいデータ」に変える作業です。
これを毎回 Excel で手作業するのは地獄なので、
Python（pandas）で「同じルールで何度でも自動で」クリーニングできるようにします。
ここでは、
1回分の「CSVクリーニングバッチ」の作り方
どんなステップをどの順番でやるか
どこでつまずきやすいか、何を意識すると“壊れない”か
を、初心者向けにじっくり解きほぐしていきます。
基本の流れ（読み込み → 型を揃える → 文字を揃える → 欠損・重複を処理 → 保存）
最小のクリーニングバッチのイメージ
まずは「とにかく一度、端から端まで流す」イメージを持ってほしいです。
例えば、次のような CSV を考えます。
- ファイル名:
raw/customers.csv - 列:
id, name, age, join_date, email - ありがちな汚れ:
- name に前後の空白や全角スペース
- age が文字列だったり「不明」が混ざる
- join_date が文字列でフォーマットバラバラ
- email がダブっている行
これを「クリーニングして別ファイルに保存する」バッチを作ります。
読み込みと基本設定（エンコード・ヘッダー・パスの扱い）
Path と絶対パスで「どこから実行しても壊れない」ようにする
バッチ化を前提にするなら、まずここをちゃんと押さえます。
from pathlib import Path
import pandas as pd
BASE_DIR = Path(__file__).resolve().parent
IN_DIR = BASE_DIR / "raw"
OUT_DIR = BASE_DIR / "clean"
IN_FILE = IN_DIR / "customers.csv"
OUT_FILE = OUT_DIR / "customers_clean.csv"
OUT_DIR.mkdir(exist_ok=True)
df = pd.read_csv(IN_FILE, encoding="utf-8")
重要なのは BASE_DIR = Path(__file__).resolve().parent です。
これで「スクリプトが置いてあるフォルダ」を基準にできるので、
cron やタスクスケジューラから実行してもパスが迷子になりません。
エンコードは UTF-8 前提が多いですが、エラーが出るなら encoding="cp932"（Windows-Excel 由来）など、実データに合わせて調整します。
型を揃える（数値・日付・カテゴリに“変換してあげる”）
数値列のクリーニング（カンマ・文字混入・空文字）
age や amount のような数値列が文字列になっていたり、「不明」「-」などが混ざっているのは超あるあるです。
pandas では「まず文字列のまま綺麗にしてから to_numeric で変換」が基本です。
df["age"] = (
df["age"]
.astype(str) # いったん文字列扱いに統一
.str.replace(",", "", regex=False) # カンマ除去
.str.strip() # 前後の空白除去
)
df["age"] = pd.to_numeric(df["age"], errors="coerce")
ここで深掘りしておきたいのは errors="coerce" です。
数値に変換できない値（”不明” など）は NaN になります。
「おかしな値は NaN に飛ばしてから、あとで欠損処理でまとめて扱う」のが安全です。
無理に「0にしておこう」といった場当たりな対処をすると、後で統計や集計をするときにひどい目に合います。
日付列のクリーニング（フォーマットの揺れ・文字列→datetime）
join_date のような日付は、Excel 由来だったり人手入力だったりで、だいたいバラバラです。
df["join_date"] = pd.to_datetime(df["join_date"], errors="coerce")
errors="coerce" を付けたことで、解釈できない日付は NaN になります。
これも後で「欠損日付」として一括で扱えます。
日付をクリーニングできたら、後続処理で「入会年」や「入会月」を作るのも定番です。
df["join_year"] = df["join_date"].dt.year
df["join_month"] = df["join_date"].dt.to_period("M").dt.to_timestamp()
文字列の揃え方（空白・全角半角・大小文字・表記揺れ）
前後の空白・全角スペースを徹底的に落とす
name や email、カテゴリ名などの文字列は、まず「余計な空白」を殺します。
def clean_str(s):
if pd.isna(s):
return s
s = str(s)
s = s.replace("\u3000", " ") # 全角スペースを半角スペースに
s = s.strip() # 前後のスペース削除
return s
df["name"] = df["name"].map(clean_str)
df["email"] = df["email"].map(clean_str)
全角スペース（\u3000）が紛れ込んでいると strip() だけでは落ちません。
実務ではここをちゃんと意識しているかどうかで、重複判定などの品質がかなり変わります。
大文字・小文字を揃える（email のようなキー系）
email やコードなどは「小文字に統一」してしまうのがよくあります。
df["email"] = df["email"].str.lower()
これで Test@Example.Com と test@example.com を同一視できるようになります。
重複チェック・JOIN・マスタ突き合わせで効きます。
表記揺れの統一（マスタを作って map する）
カテゴリ名などで、
- “男性”, “男”, “M”
- “女性”, “女”, “F”
のような揺れがある場合は、マスタを作って map します。
gender_map = {
"男": "M",
"男性": "M",
"Male": "M",
"女": "F",
"女性": "F",
"Female": "F"
}
df["gender"] = df["gender"].map(clean_str).map(gender_map).fillna("UNKNOWN")
ここでのポイントは、
- まず clean_str で空白や全角を落としてから map
- 不明な値は “UNKNOWN” などにまとめる（NaN のままにしない場合）
という流れです。
欠損値（NaN）の扱い方（消すのか、埋めるのか、フラグを立てるのか）
まずは「欠損状況をざっくり把握」する
クリーニングの前半で、「数値変換できなかったもの」「日付にできなかったもの」はすでに NaN になっています。
ここで一度、欠損状況を見ておきます。
print(df.isna().sum())
print((df.isna().mean() * 100).round(1))
列ごとの欠損数・欠損率を見て、
- ほとんど空っぽの列は捨ててしまう
- 一部だけ欠損なら補完 or 除外
といった判断をします。
シンプルな補完方法（fillna）
典型的なのは「平均・中央値・最頻値」で埋めるパターンです。
df["age"] = df["age"].fillna(df["age"].median())
df["join_date"] = df["join_date"].fillna(pd.Timestamp("2000-01-01"))
df["email"] = df["email"].fillna("unknown@example.com")
ただし、闇雲に埋めるのではなく、
- そもそもその列が何を意味しているか
- その値で埋めると分析や下流処理でどう見なされるか
を考えてから決める必要があります。
例えば age を中央値で埋めるのはそこそこ自然ですが、
join_date をでたらめに埋めると「いつ入会したか」という意味が崩れるので、
別列 join_date_missing_flag を作って「欠損だったかどうか」を残しておくのがよくある手です。
df["join_date_missing"] = df["join_date"].isna().astype(int)
df["join_date"] = df["join_date"].fillna(pd.Timestamp("2000-01-01"))
重複と異常値（おかしな行を「見つけて」「どうするか決める」）
完全重複行の削除
まったく同じ行が重複している場合は、単純に drop_duplicates で落とせます。
df = df.drop_duplicates()
ただし、多くの場合「キーだけ同じで他の列が違う」パターンもあります。
そういうときは、重複チェックのキーを明示します。
dups = df[df.duplicated(subset=["email"], keep=False)]
print(dups)
どれを残すかはビジネスルール次第です。
- 最新の join_date を残す
- 金額の大きい方を残す
- 人間が決定するためのリストを出す
など、ルールを決めたら、そのロジックを関数化してあげます。
異常値の検出（範囲チェック）
年齢が 0 や 150 になっているような明らかな異常もよく出ます。
まずは条件で抽出して「どう扱うか」を考えます。
abnormal_age = df[(df["age"] < 0) | (df["age"] > 100)]
print(abnormal_age)
扱い方としては、
- NaN にして「欠損」と同じ扱いにする
- 上限・下限でクリップする（例えば 0 未満は 0、100 超は 100 にする）
- その行自体を除外する
などがあります。
df.loc[(df["age"] < 0) | (df["age"] > 100), "age"] = None
df["age"] = df["age"].fillna(df["age"].median())
重要なのは「ルールをコードに明文化する」ことです。
Excel 手作業だと「なんとなく良さそう」にやってしまいがちですが、Python にするとルールが明示されます。
一連のクリーニングを「バッチ」としてまとめる
関数に分解して“読みやすいクリーニングレシピ”にする
ここまでの処理を、1ファイルのバッチにまとめてみます。
# clean_customers_batch.py
import logging
from pathlib import Path
from datetime import datetime
import pandas as pd
BASE_DIR = Path(__file__).resolve().parent
IN_DIR = BASE_DIR / "raw"
OUT_DIR = BASE_DIR / "clean"
LOG_FILE = BASE_DIR / "logs" / "clean_customers.log"
IN_FILE = IN_DIR / "customers.csv"
def setup_logging():
LOG_FILE.parent.mkdir(exist_ok=True)
logging.basicConfig(
filename=LOG_FILE,
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
def clean_str(s):
if pd.isna(s):
return s
s = str(s)
s = s.replace("\u3000", " ")
s = s.strip()
return s
def load_data():
logging.info(f"読み込み: {IN_FILE}")
df = pd.read_csv(IN_FILE, encoding="utf-8")
logging.info(f"行数: {len(df)}")
return df
def clean_types(df):
df["age"] = (
df["age"]
.astype(str)
.str.replace(",", "", regex=False)
.str.strip()
)
df["age"] = pd.to_numeric(df["age"], errors="coerce")
df["join_date"] = pd.to_datetime(df["join_date"], errors="coerce")
return df
def clean_strings(df):
df["name"] = df["name"].map(clean_str)
df["email"] = df["email"].map(clean_str).str.lower()
return df
def handle_missing(df):
df["join_date_missing"] = df["join_date"].isna().astype(int)
df["age"] = df["age"].fillna(df["age"].median())
df["join_date"] = df["join_date"].fillna(pd.Timestamp("2000-01-01"))
df["email"] = df["email"].fillna("unknown@example.com")
return df
def handle_duplicates(df):
before = len(df)
df = df.drop_duplicates()
after = len(df)
logging.info(f"重複削除: {before} -> {after}")
return df
def handle_outliers(df):
cond = (df["age"] < 0) | (df["age"] > 100)
n_out = cond.sum()
if n_out > 0:
logging.info(f"年齢の異常値を {n_out} 件 NaN に変更")
df.loc[cond, "age"] = None
df["age"] = df["age"].fillna(df["age"].median())
return df
def save_data(df):
OUT_DIR.mkdir(exist_ok=True)
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
out_path = OUT_DIR / f"customers_clean_{ts}.csv"
df.to_csv(out_path, index=False, encoding="utf-8-sig")
logging.info(f"保存完了: {out_path}")
def run_job():
df = load_data()
df = clean_types(df)
df = clean_strings(df)
df = handle_missing(df)
df = handle_duplicates(df)
df = handle_outliers(df)
save_data(df)
def main():
logging.info("顧客CSVクリーニングバッチ開始")
try:
run_job()
logging.info("顧客CSVクリーニングバッチ正常終了")
except Exception as e:
logging.exception(f"顧客CSVクリーニングバッチ異常終了: {e}")
raise
if __name__ == "__main__":
setup_logging()
main()
このように、処理を小さな関数に分けると
- 「何をやっているバッチか」が一目で分かる
- 一部だけロジックを変えたいときに差し替えやすい
- テストもしやすい
というメリットがあります。
まとめ（「型を揃える → 文字を揃える → 欠損と重複を整理する」をレシピ化する）
CSV のクリーニングを Python で自動化するときのキモは、
- Path と絶対パスで「どこから実行しても同じ結果」にする
- 数値・日付をまず正しい型に変換し、変換できないものは NaN に飛ばす
- 文字列の空白・全角スペース・大小文字・表記揺れをそろえる
- 欠損・重複・異常値は「どう扱うか」をルールとしてコードに書く
- 一連のステップを関数に分解して “クリーニングレシピ” として残す
この流れを一度きちんと作ってしまえば、
新しい CSV が来ても「同じコードを回せば、毎回同じ品質でクリーンになる」状態が作れます。
