data-cleaning

データクリーニング

このスキルを使うと、AI エージェントは生データセットを体系的にクリーニングおよび前処理し、分析にすぐ使える状態にできます。欠損値、重複レコード、データ型の不一致、書式の不統一、外れ値の処理、正規化に対応します。また、検証スキーマを適用して、継続的なデータ品質を確保できます。主なツールチェーンは pandas で、高度な検証には pyjanitor と great_expectations を利用します。

ワークフロー

1. 生データを取り込み、プロファイリングします。 データセットを読み込み、すぐに品質レポートを作成します。列ごとの null 数を数え、重複行を特定し、データ型が想定スキーマと一致しているかを確認し、型が混在する列にフラグを付けます。このプロファイルは、その後のすべてのクリーニング判断の基礎となります。

2. 欠損値を処理します。 データ型と欠損パターンに応じて、列ごとに処理方法を適用します。欠損率が 5% 未満の数値列には、中央値による補完を使用します。カテゴリ列には、最頻値または専用の「Unknown」カテゴリを使用します。欠損率が 40% を超える列には削除候補としてフラグを付け、削除する前にユーザーに確認します。

3. 重複を削除し、競合を解決します。 完全一致の重複と、空白や大文字・小文字の違いのみがある行などの近似重複を特定します。完全一致の重複では、最初の出現を残します。近似重複には、設定可能なしきい値を使ったファジーマッチングを適用し、値が競合する場合は、新しさまたは完全性に基づいて統合します。

4. データ型を修正し、書式を標準化します。 列を意図した型に変換します。日付文字列を datetime オブジェクトに解析し、数値文字列を浮動小数点数に変換し、カテゴリ値を標準形に正規化します。電話番号、郵便番号、通貨表記などの書式を標準化します。

5. 外れ値を検出して処理します。 対称分布には IQR 法（1.5x）、正規分布データには z スコアを使用します。境界値で上限・下限を設定する（ウィンザー化）、後で補完できるよう null に置き換える、元の値を保持したまま注釈を付けるだけのモード、という 3 つの処理方法を提示します。

6. クリーニング後の出力を検証します。 クリーニング済みデータセットに検証ルールを適用します。null 不可制約、範囲チェック、一意性制約、参照整合性を確認します。残っている違反を報告し、適用したすべての変換を記録したクリーニングログとともに、クリーンなデータセットを保存します。

サポート対象のテクノロジー

• pandas — 中核となるデータ操作と型変換
• pyjanitor — クリーニング操作をメソッドチェーンで簡潔に記述
• great_expectations — スキーマ検証とデータ品質チェック
• fuzzywuzzy — 近似重複の検出に使うファジー文字列マッチング
• numpy — 外れ値検出の数値演算

使い方

生データセットのファイルパスをエージェントに渡します。必要に応じて、想定される列の型、有効範囲、一意性制約を指定したスキーマ定義も渡します。エージェントは、クリーニング済みファイルと変換ログを作成します。

例

例 1：pandas を使った不整合のある CSV のクリーニング

import pandas as pd
import numpy as np

# Load raw data
df = pd.read_csv("messy_orders.csv")
print(f"Raw shape: {df.shape}")  # (2340, 8)
print(df.isnull().sum())
# order_id         0
# customer_name   12
# email           45
# order_date      18
# amount          23
# status           0
# region          67
# discount         0

# 1. Fix data types — order_date has mixed formats
df["order_date"] = pd.to_datetime(df["order_date"], format="mixed", dayfirst=False)
df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

# 2. Handle missing values
df["customer_name"] = df["customer_name"].fillna("Unknown")
df["email"] = df["email"].fillna("[email protected]")
df["amount"] = df["amount"].fillna(df["amount"].median())
df["region"] = df["region"].fillna(df["region"].mode()[0])
df["order_date"] = df["order_date"].fillna(method="ffill")

# 3. Remove duplicates
before = len(df)
df = df.drop_duplicates(subset=["order_id"], keep="first")
print(f"Removed {before - len(df)} duplicate orders")  # Removed 34 duplicate orders

# 4. Standardize categorical values
df["status"] = df["status"].str.strip().str.lower().replace({
    "shipped": "shipped", "ship": "shipped",
    "cancelled": "cancelled", "canceled": "cancelled",
    "pending": "pending", "pend": "pending"
})
df["region"] = df["region"].str.strip().str.title()

# 5. Outlier treatment — cap amounts at IQR bounds
Q1 = df["amount"].quantile(0.25)
Q3 = df["amount"].quantile(0.75)
IQR = Q3 - Q1
lower, upper = Q1 - 1.5 * IQR, Q3 + 1.5 * IQR
df["amount"] = df["amount"].clip(lower=lower, upper=upper)

print(f"Clean shape: {df.shape}")  # (2306, 8)
df.to_csv("clean_orders.csv", index=False)

例 2：スキーマ適用によるデータ検証パイプライン

import great_expectations as gx

context = gx.get_context()

# Define a validation suite
suite = context.add_expectation_suite("orders_validation")

# Add expectations
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount", min_value=0.01, max_value=50000.00
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeInSet(
        column="status", value_set=["pending", "shipped", "delivered", "cancelled"]
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToMatchRegex(
        column="email", regex=r"^[^@]+@[^@]+\.[^@]+$"
    )
)

# Run validation against cleaned data
results = context.run_validation(suite, batch=gx.read_csv("clean_orders.csv"))

print(f"Success: {results.success}")
print(f"Passed: {results.statistics['successful_expectations']}/{results.statistics['evaluated_expectations']}")
# Success: True
# Passed: 5/5

ベストプラクティス

• 変換を行う前に、必ず生データのコピーを保存します。クリーニングは破壊的ではなく、再現可能であるべきです。
• 監査可能なクリーニング履歴を作成するため、変換内容を件数とともにすべて記録します（例：「amount の null 23 件を中央値 312.45 で補完」）。
• 機械的な既定値よりも、ドメイン知識に基づく補完を優先します。ある列の欠損パターンがランダムでない（MNAR）場合は、ユーザーに相談します。
• 早い段階から頻繁に検証します。最後だけでなく、各クリーニング段階の後にスキーマチェックを実行します。
• クリーニングは反復的に行います。最初の処理で明らかな問題を検出できますが、後続の分析で新たな問題が見つかることもよくあります。
• pd.to_numeric と pd.to_datetime では errors="coerce" を使い、変換に失敗した場合に処理を停止するのではなく、NaN として検出します。

エッジケース

• 完全に空の列。 列の値がすべて null の場合は、情報がない状態で補完を試みるのではなく、自動的に削除して警告を記録します。
• 列名の重複。 Pandas では列名の重複が黙って許容されます。読み込み時に重複を検出し、操作を行う前にサフィックス（_1、_2）を付けて名前を変更します。
• エンコーディングの問題。 read_csv が UnicodeDecodeError を発生させた場合は、encoding="latin-1"、次に encoding="cp1252" で再試行し、成功したエンコーディングを記録します。
• 複数の形式が混在する日付列。 1 つの列に「2024-01-15」、「01/15/2024」、「Jan 15, 2024」が含まれる場合は、pd.to_datetime(col, format="mixed") を使用し、スポットチェックで解析結果を検証します。
• 通貨記号付きの文字列として保存された数値列。 型変換の前に、$、€、カンマ、空白を除去します：df["price"].str.replace(r"[$€,\s]", "", regex=True).astype(float)。