DuckDB×PythonでCSV品質チェックをやってみた：欠損・重複・異常値をSQLで確認する

CSVデータを分析や可視化に使う前に、欠損値や重複、異常値が含まれていないか確認したい場面があります。

もちろん、PythonのpandasだけでもCSVの品質チェックはできます。例えば、欠損値の確認や重複チェックはDataFrameを df とした場合、次のように書けます。

df.isnull().sum()
df.duplicated(subset=["inquiry_id"]).sum()
df[df["response_minutes"] > 1440]

一方で、データ分析やデータ基盤の現場では、SQLでデータを確認する機会も多くあります。DuckDBを使うと、ローカルPC上でCSVを読み込み、SQLで手軽に品質チェックを実行できます。

この記事では、DuckDBを使って、問い合わせログを模した架空CSVの品質チェックを行います。

なお、この記事では「問題データを検出するところ」までを扱います。検出したデータの修正、クレンジング処理の自動化、本番運用の設計までは扱いません。

DuckDBとは

DuckDBは、ローカル環境で手軽にSQLを実行できる分析向けデータベースです。

Pythonから利用でき、CSVやParquetなどのファイルを直接読み込んでSQLを実行できます。サーバー型のデータベースを用意しなくても試せるため、手元のCSVを少し確認したいときに便利です。

今回のような小さな検証では、DuckDBをメモリ上で起動して使います。

実行環境

筆者は以下の環境で実行しました。

ライブラリは、仮想環境内でインストールしています。

PowerShell

python -m pip install duckdb pandas

インストール後、以下のコマンドでDuckDBのバージョンを確認できます。

PowerShell

python -c "import duckdb; print(duckdb.__version__)"

ModuleNotFoundError: No module named 'duckdb' が表示される場合は、VS Codeで選択しているPython環境に duckdb がインストールされていない可能性があります。

その場合は、VS Code右下のPython環境表示を確認し、作成した仮想環境 .venv が選択されているか確認します。

コードの実行方法

この記事では、duckdb_csv_quality_check.py というPythonファイルを作成し、以降のコードを上から順番に書いて実行する想定で進めます。

作業フォルダの例は以下です。

duckdb_csv_quality_check/
├── duckdb_csv_quality_check.py

duckdb_csv_quality_check.py を保存したら、ターミナルでそのファイルがあるフォルダに移動し、以下のコマンドで実行します。

PowerShell

python duckdb_csv_quality_check.py

このあと作成する inquiry_logs.csv や inquiry_logs.parquet は、基本的にこのPythonファイルと同じフォルダに作成されます。

今回作成する架空CSV

今回は、問い合わせログをイメージした架空CSVを作成します。

列は以下のようにします。

品質チェックの動きを確認しやすくするために、あえて以下の問題データも混ぜます。

Pythonで架空CSVを作成する

まず、Pythonで100件分の通常データを作成します。

以降のコードは、先ほど作成した duckdb_csv_quality_check.py に上から順番に書いていきます。

import duckdb
import pandas as pd
from datetime import datetime, timedelta

categories = ["アクセス申請", "エラー・トラブル", "使い方ガイド", "データ確認"]
priorities = ["high", "middle", "low"]
statuses = ["open", "in_progress", "resolved"]
channels = ["Teams", "Web", "Email"]
ratings = ["good", "neutral", "bad"]

rows = []
base_time = datetime(2026, 6, 1, 9, 0, 0)

for i in range(1, 101):
    rows.append({
        "inquiry_id": f"INQ-{i:04d}",
        "created_at": base_time + timedelta(minutes=i * 15),
        "category": categories[i % len(categories)],
        "priority": priorities[i % len(priorities)],
        "status": statuses[i % len(statuses)],
        "channel": channels[i % len(channels)],
        "response_minutes": (i * 3) % 120,
        "resolution_minutes": (i * 11) % 600,
        "rating": ratings[i % len(ratings)],
        "comment": "架空データです"
    })
    
df = pd.DataFrame(rows)

次に、品質チェックで検出するための問題データを追加します。

pandasの行番号は0から始まります。そのため、例えば df.loc[3] はCSV上の4件目のデータを指します。

# 欠損値を作る
df.loc[3, "category"] = None
df.loc[10, "rating"] = None

# 異常値を作る
df.loc[20, "response_minutes"] = 9999
df.loc[25, "resolution_minutes"] = -30

# 想定外のカテゴリ値を作る
df.loc[30, "status"] = "done"
df.loc[35, "priority"] = "urgent"

# 重複データを作る
duplicate_row = df.loc[5].copy()
df = pd.concat([df, pd.DataFrame([duplicate_row])], ignore_index=True)

ここでは、以下の問題を意図的に作っています。

• category と rating に欠損値を入れる
• response_minutes に極端に大きい値を入れる
• resolution_minutes にマイナス値を入れる
• status に想定外の done を入れる
• priority に想定外の urgent を入れる
• INQ-0006 の行を複製して、重複データを作る

最後に、作成したDataFrameをCSVファイルとして保存します。

df.to_csv("inquiry_logs.csv", index=False, encoding="utf-8-sig")

print("CSVを作成しました")
print(df.head())

encoding="utf-8-sig" を指定しているのは、ExcelなどでCSVを開いたときに日本語が文字化けしにくくするためです。

ここまでで、inquiry_logs.csv というCSVファイルが作成されます。

作成されるCSVの先頭5行は、以下のようなイメージです。

DuckDBでCSVを読み込む

次に、DuckDBに接続します。

con = duckdb.connect()

duckdb.connect() を引数なしで実行すると、メモリ上のDuckDBに接続できます。今回のような小さな検証であれば、この形で十分です。

次に、作成したCSVをDuckDBのテーブルとして読み込みます。

con.execute("""
CREATE OR REPLACE TABLE inquiry_logs AS
SELECT *
FROM read_csv('inquiry_logs.csv', header = true)
""")

read_csv を使うと、CSVを読み込んでSQLから扱えます。今回はCSVの先頭行に列名があるため、header = true を指定しています。

また、inquiry_logs.csv は相対パスで指定しています。この場合、Pythonを実行している作業フォルダから見て inquiry_logs.csv が存在する必要があります。この記事の手順どおりに進めていれば、Pythonファイルと同じフォルダにCSVが作成されます。

DuckDBの read_csv はCSVの区切り文字や列の型などを自動推論できます。今回のような検証では自動推論に任せますが、実データで型を厳密に扱いたい場合は、列の型を明示することも検討します。

CREATE OR REPLACE TABLE を使っているため、同じ名前のテーブルがすでに存在していても作り直せます。何度もコードを実行し直す検証では便利です。

SQLで品質チェックを行う

ここからは、DuckDBでSQLを実行してCSVの品質を確認します。

なお、この記事ではPythonからDuckDBにSQLを渡して実行しています。SQL部分だけを見ると、通常のSQLによる品質チェックと同じ考え方で読めます。

今回確認する内容は以下です。

チェック１：行数を確認する

まず、読み込んだデータの行数を確認します。

print("\n--- 行数確認 ---")
print(con.execute("""
SELECT count(*) AS row_count
FROM inquiry_logs
""").df())

実行結果は以下です。

--- 行数確認 ---
   row_count
         101

今回は、100件の架空データを作成したあと、重複確認用に1行追加しました。そのため、行数は101件になります。

最初に行数を確認しておくと、CSVが想定どおり読み込まれているか確認できます。

チェック２：欠損値を確認する

次に、重要な列に欠損値がないか確認します。

print("\n--- 欠損値チェック ---")
print(con.execute("""
SELECT
    count(*) AS total_rows,
    sum(CASE WHEN inquiry_id IS NULL THEN 1 ELSE 0 END) AS missing_inquiry_id,
    sum(CASE WHEN category IS NULL THEN 1 ELSE 0 END) AS missing_category,
    sum(CASE WHEN rating IS NULL THEN 1 ELSE 0 END) AS missing_rating,
    sum(CASE WHEN response_minutes IS NULL THEN 1 ELSE 0 END) AS missing_response_minutes,
    sum(CASE WHEN resolution_minutes IS NULL THEN 1 ELSE 0 END) AS missing_resolution_minutes,
    sum(CASE WHEN status IS NULL THEN 1 ELSE 0 END) AS missing_status,
    sum(CASE WHEN priority IS NULL THEN 1 ELSE 0 END) AS missing_priority
FROM inquiry_logs
""").df())

このSQLでは、各列が NULL の場合に1を足しています。

例えば、category IS NULL の場合に1、それ以外の場合に0を返し、その合計を取ることで欠損件数を確認しています。

実行結果は以下です。

--- 欠損値チェック ---
total_rows  missing_inquiry_id  missing_category  missing_rating  missing_response_minutes  missing_resolution_minutes  missing_status  missing_priority
       101                 0.0               1.0             1.0                       0.0                         0.0             0.0               0.0

category と rating に、それぞれ1件ずつ欠損値があることが分かります。

category が欠損していると、問い合わせカテゴリ別の集計に影響します。rating が欠損していると、評価別の分析に影響します。

このように、分析前に欠損値を確認しておくことで、後続の集計結果を確認しやすくなります。

なお、DuckDBでは以下のように FILTER を使って欠損件数を数えることもできます。

count(*) FILTER (category IS NULL)

この記事では、SQL初心者にも条件分岐の考え方が分かりやすいように、CASE WHEN を使っています。

チェック３：重複データを確認

次に、問い合わせIDが重複していないか確認します。

print("\n--- 重複チェック ---")
print(con.execute("""
SELECT
    inquiry_id,
    count(*) AS duplicate_count
FROM inquiry_logs
GROUP BY inquiry_id
HAVING count(*) > 1
ORDER BY duplicate_count DESC
""").df())

このSQLでは、inquiry_id ごとに件数を数えています。

HAVING count(*) > 1 を指定することで、2件以上存在する問い合わせIDだけを抽出できます。

実行結果は以下です。

--- 重複チェック ---
  inquiry_id  duplicate_count
    INQ-0006                2

INQ-0006 が2件存在していることが分かります。

問い合わせIDやチケットIDのように一意であるべき列は、重複チェックの対象にするとよいです。重複したまま集計すると、同じ問い合わせを二重に数えてしまう可能性があります。

チェック４：異常値を確認する

次に、応答時間や解決時間に異常値がないか確認します。

今回は、以下の条件を異常値候補とします。

1440分 は1日、10080分 は7日です。

print("\n--- 異常値チェック ---")
print(con.execute("""
SELECT
    inquiry_id,
    response_minutes,
    resolution_minutes
FROM inquiry_logs
WHERE response_minutes < 0
   OR response_minutes > 1440
   OR resolution_minutes < 0
   OR resolution_minutes > 10080
ORDER BY inquiry_id
""").df())

実行結果は以下です。

--- 異常値チェック ---
  inquiry_id  response_minutes  resolution_minutes
    INQ-0021              9999                 231
    INQ-0026                78                 -30

INQ-0021 は、response_minutes が 9999 になっています。初回応答までの時間としては極端に大きいため、異常値候補として確認できます。

また、INQ-0026 は、resolution_minutes が -30 になっています。解決までの時間がマイナスになることは通常考えにくいため、こちらも異常値候補として確認できます。

ただし、異常値の基準は業務ルールによって変わります。問い合わせ対応のルールによっては、1日以上かかっても異常ではない場合があります。逆に、数時間以内の対応が必須であれば、もっと短い時間をしきい値にする必要があります。

そのため、今回の条件はあくまで検証用の条件です。

チェック５：想定外のstatusの値を確認する

次に、status にどのような値が入っているか確認します。

print("\n--- statusの値確認 ---")
print(con.execute("""
SELECT
    status,
    count(*) AS count
FROM inquiry_logs
GROUP BY status
ORDER BY count DESC, status
""").df())

実行結果は以下です。

--- statusの値確認 ---
        status  count
          open     34
   in_progress     33
      resolved     33
          done      1

open、in_progress、resolved のほかに、done が1件存在していることが分かります。

done は今回の想定値に含めていません。

次に、想定外の status だけを抽出します。

print("\n--- 想定外statusチェック ---")
print(con.execute("""
SELECT
    inquiry_id,
    status
FROM inquiry_logs
WHERE status IS NULL
   OR status NOT IN ('open', 'in_progress', 'resolved')
""").df())

実行結果は以下です。

--- 想定外statusチェック ---
  inquiry_id status
    INQ-0031   done

INQ-0031 の status に、想定外の done が入っていることが分かります。

ここでは status IS NULL OR status NOT IN (...) と書いています。NOT IN だけでは NULL の値を抽出できないため、実データの品質チェックでは IS NULL もあわせて確認すると安全です。

このような値が混ざると、ステータス別集計の分類がぶれる可能性があります。例えば、resolved と done が同じ意味で使われている場合でも、値が分かれていると別カテゴリとして集計されます。

そのため、事前に想定値を決めて、想定外の値がないか確認することが重要です。

チェック６：想定外のpriorityを確認する

同じように、priority も確認します。

priority の想定値は、以下の3つとします。

以下のSQLで、想定外の値を抽出します。

print("\n--- 想定外priorityチェック ---")
print(con.execute("""
SELECT
    inquiry_id,
    priority
FROM inquiry_logs
WHERE priority IS NULL
   OR priority NOT IN ('high', 'middle', 'low')
""").df())

実行結果は以下です。

--- 想定外priorityチェック ---
  inquiry_id priority
    INQ-0036   urgent

INQ-0036 の priority に、想定外の urgent が入っていることが分かります。

もし urgent を正式な優先度として扱うのであれば、想定値に追加する必要があります。一方で、入力ミスとして扱うのであれば、high などの既存カテゴリに修正する必要があります。

補足：DuckDBでParquet形式に出力する手順

ここまでで、CSVに含まれる問題データをSQLで確認できました。

最後に補足として、DuckDBの COPY 文を使って、読み込んだデータをParquet形式で出力してみます。

con.execute("""
COPY (
    SELECT *
    FROM inquiry_logs
) TO 'inquiry_logs.parquet' (FORMAT parquet)
""")

print("\nParquetファイルを作成しました")

これで、inquiry_logs.parquet が作成されます。

Parquetは列指向のファイル形式で、分析用途で扱いやすい場面があります。

ただし、今回のコードでは、検出した問題データを修正しているわけではありません。そのため、ここで作成するParquetファイルは「品質チェック済みのきれいなデータ」ではなく、「DuckDBで読み込んだデータを別形式で出力する例」として扱います。

つまり、このセクションは品質チェックそのものではなく、DuckDBで読み込んだデータを別形式に変換する補足です。

作成したParquetファイルの件数も確認します。

print("\n--- Parquet確認 ---")
print(con.execute("""
SELECT count(*) AS parquet_row_count
FROM read_parquet('inquiry_logs.parquet')
""").df())

実行結果は以下です。

--- Parquet確認 ---
   parquet_row_count
                 101

CSVから読み込んだ行数と、Parquetファイルから読み込んだ行数が同じ101件であることを確認できました。

最後に、DuckDBの接続を閉じます。

con.close()

今回使ったSQLの考え方

今回の品質チェックでは、以下のようなSQLを使いました。

どれも基本的なSQLですが、組み合わせることでCSVの品質チェックに使えます。

DuckDBでCSV品質チェックを行うときの注意点

DuckDBで品質チェックを行うときは、チェック条件を先に決めておくことが重要です。

例えば、今回の異常値チェックでは、response_minutes > 1440 を異常値候補としました。しかし、どの値を異常とするかは業務によって変わります。

問い合わせ対応のルールによっては、1日以上かかっても異常ではない場合があります。逆に、数時間以内の対応が必須であれば、もっと短い時間をしきい値にする必要があります。

また、status や priority のようなカテゴリ値も、あらかじめ想定値を決めておく必要があります。想定値が決まっていないと、どの値を不整合として扱うか判断できません。

そのため、DuckDBは「問題候補を見つける道具」として使い、最終的な判断は業務ルールに合わせて行う必要があります。

まとめ

この記事では、DuckDBを使って、架空CSVに含まれる欠損値、重複、異常値、想定外のカテゴリ値をSQLで確認しました。

PythonだけでもCSVの品質チェックは可能です。一方で、DuckDBを使うと、CSVをテーブルのように扱いながら、SQLで一貫したチェックロジックを書けます。

今回確認した内容は以下です。

• CSVの行数を確認する
• 必要な列の欠損値を確認する
• 一意であるべきIDの重複を確認する
• 応答時間や解決時間の異常値候補を確認する
• status や priority のNULLや想定外の値を確認する
• 補足として、読み込んだデータをParquet形式に出力する

データ分析の前にこのような確認を行うことで、集計結果の誤りや分類のぶれに気づきやすくなります。

小規模で単発の確認であればpandasだけでも十分です。一方で、SQLでチェックロジックを残したい場合や、後から別のデータにも同じ確認を適用したい場合は、DuckDBが便利な選択肢になります。

参考

• DuckDB公式ドキュメント｜Python API
  https://duckdb.org/docs/current/clients/python/overview
• DuckDB公式ドキュメント｜CSV Import
  https://duckdb.org/docs/current/data/csv/overview
• DuckDB公式ドキュメント：Reading and Writing Parquet Files
  https://duckdb.org/docs/current/data/parquet/overview