- カテゴリー
- Python
- 公開日
- 2026.09.09
Contents
はじめに
月別のCSVをまとめようとしたら、先月と今月で列の順番が違う。途中からメモ列が増え、金額に「未確定」が混じっている。こうしたデータは、読み込めただけでは正しく集計できたとは判断できません。
この記事では、Pythonの関数とCSVの基本を学んだ人向けに、DuckDBで3か月分の売上CSVを統合するスクリプトを作ります。列名でデータをそろえ、金額・必須項目・注文IDの重複を確認し、問題がない場合だけ集計します。SQLはデータを選択・集計するための言語で、使う構文はコードの後で説明します。
DuckDBはPythonのプログラム内から使える分析向けのデータベースです。今回は別のデータベースサーバーを立てず、端末内のメモリで処理します。元のCSVは書き換えません。
作るものと入力のルール
作るのは、フォルダーを指定するとファイル別の件数・合計と、カテゴリ別の集計を表示するコマンドです。CSVを行方向にまとめ、同じ名前の列を対応づけます。注文IDをキーに別の表を横へ結合するJOINではありません。
教材には架空の売上を使います。3ファイル6行の合計は6,500円です。入力ルールを先に決めます。
| 項目 | この教材のルール |
|---|---|
| 文字コードと区切り | UTF-8またはUTF-8 BOM付き、カンマ区切り、先頭行はヘッダー |
| 必須列 | order_id、category、amount。すべてのファイルに必要 |
| 任意列 | memo。列自体がない月も受け付ける |
| 列名 | 表記は上記の小文字と完全一致。未知の列・重複した列名は中止 |
| 金額 | 円単位の整数。返品の負数や先頭の+は許可。小数・桁区切りは不可 |
| 注文ID | 前後の空白を除き、全ファイルで一意。同じ注文の複数明細は扱わない |
| 入力フォルダー | 読みたいCSVだけを置く。直下の*.csvが対象 |
ファイル名を月のラベルとして表示します。日付列から月を計算したり、ファイル名と実際の取引月が合っているかを判定したりする機能はありません。
仕組み：列をそろえる処理と、値を確かめる処理を分ける
DuckDBのunion_by_name=trueは、複数ファイルの列を名前で対応づける指定です。あるファイルにない列にはNULLが入ります。NULLはSQLで値がないことを表します。詳細は公式の列構成統合ガイドで確認できます。
ただし、「amountとamoutは同じ意味だろう」と推測して直す機能ではありません。今回はPythonで各ファイルのヘッダーを確認し、必須列の不足や未知の列を先に止めます。
金額の値は、all_varchar=trueでいったん文字列として読みます。注文IDの001を数値の1に変えず、金額の「未確定」も確認対象として残すためです。その後、整数の表記に合うかを調べてから数値に変換します。
処理は次の順です。
- 各CSVのヘッダーを確認する。
- 列名でそろえて、一時テーブル
rawへ読み込む。 - 値を整え、金額の変換結果を
checkedへ保存する。 - 不正値と注文IDの重複があれば、集計せずに中止する。
- ファイル別・カテゴリ別・全体の件数と金額を表示する。
必要な環境
掲載コードはWindows 11、PowerShell、Python 3.13.9、DuckDB 1.5.5で実行しました。追加の有料サービスやpandasは使いません。Pythonのインストールと仮想環境についてはvenv・pipの入門記事で補えます。
DuckDB v2のalphaも公式に試用案内されていますが、この教材は安定版1.5.5に固定します。列名でのCSV統合はv2で初めて使える機能ではありません。
手順1：作業フォルダーと仮想環境を作る
PowerShellで、まだ存在しない作業用フォルダーを作ります。同名のフォルダーがある場合は別名を選んでください。
mkdir monthly-csv-lab
cd monthly-csv-lab
py -3.13 -m venv .venv
& .\.venv\Scripts\python.exe -m pip install duckdb==1.5.5
mkdir data
& .\.venv\Scripts\python.exe -c "import sys, duckdb; print(sys.version.split()[0]); print(duckdb.__version__)"この環境では次のバージョンが表示されます。
3.13.9
1.5.5py -3.13が見つからない場合は、インストールしたPython 3.13の実行ファイルをフルパスで指定してください。仮想環境を作った後は、上記のようにその中のpython.exeを直接使います。activateの操作は不要です。パッケージ導入時にはインターネット接続が必要です。
完成時の配置は次のとおりです。
monthly-csv-lab/
.venv/
merge_monthly.py
data/
2026-06.csv
2026-07.csv
2026-08.csv手順2：列順の違う3つのCSVを用意する
エディターで以下のファイルを作り、UTF-8で保存します。エクスプローラーで拡張子を表示し、.csv.txtになっていないことも確認してください。
data/2026-06.csv
6月にはmemo列がありません。
order_id,category,amount
001,書籍,1200
002,文房具,500data/2026-07.csv
7月は金額が先頭に移動し、memo列が増えています。注文004のメモは空欄です。
amount,order_id,category,memo
1800,003,書籍,追加注文
700,004,文房具,data/2026-08.csv
8月はカテゴリが先頭で、返品の金額も含みます。
category,memo,order_id,amount
書籍,返品,005,-200
食品,試食会,006,2500CSVの列順を手作業で並べ替える必要はありません。6月のメモ列の欠落と、7月の空欄は、読み込み後にはどちらもNULLになります。両者を厳密に区別したい業務では、ヘッダーの情報も別途保持する設計が必要です。
手順3：統合と検算のスクリプトを作る
monthly-csv-lab直下にmerge_monthly.pyを作り、次のコードを保存してください。SQLはPythonの三重引用符で囲んでいます。
import argparse
import csv
import sys
from pathlib import Path
import duckdb
def merge(folder):
files = sorted(folder.glob("*.csv"))
if not files:
raise ValueError("入力フォルダーにCSVがありません")
required = {"order_id", "category", "amount"}
allowed = required | {"memo"}
for path in files:
with path.open(encoding="utf-8-sig", newline="") as file:
header = next(csv.reader(file, strict=True), [])
if len(header) != len(set(header)):
raise ValueError(f"{path.name}: ヘッダーが重複しています")
if not required <= set(header) or not set(header) <= allowed:
raise ValueError(f"{path.name}: 必須列の不足または未知の列があります")
with duckdb.connect(":memory:") as con:
con.execute("""
CREATE TABLE raw AS
SELECT * FROM read_csv(
?, header=true, delim=',', quote='"', escape='"',
union_by_name=true, all_varchar=true, filename=true,
strict_mode=true, null_padding=false, ignore_errors=false
)
""", [[str(path.resolve()) for path in files]])
if "memo" not in [row[0] for row in con.execute("DESCRIBE raw").fetchall()]:
con.execute("ALTER TABLE raw ADD COLUMN memo VARCHAR")
con.execute("""
CREATE TABLE checked AS
SELECT filename, trim(order_id) AS order_id,
trim(category) AS category, amount AS amount_text, memo,
CASE WHEN regexp_full_match(trim(amount), '[+-]?[0-9]+')
THEN TRY_CAST(trim(amount) AS BIGINT)
END AS amount_yen
FROM raw
""")
invalid = con.execute("""
SELECT filename, order_id, category, amount_text
FROM checked
WHERE order_id IS NULL OR order_id = ''
OR category IS NULL OR category = '' OR amount_yen IS NULL
ORDER BY filename, order_id
""").fetchall()
if invalid:
details = "\n".join(
f"{Path(f).name}: order_id={i!r}, category={c!r}, amount={a!r}"
for f, i, c, a in invalid
)
raise ValueError("入力値を確認してください:\n" + details)
duplicates = con.execute("""
SELECT order_id, count(*) FROM checked
GROUP BY order_id HAVING count(*) > 1 ORDER BY order_id
""").fetchall()
if duplicates:
raise ValueError(f"order_idが重複しています: {duplicates}")
count = con.execute("SELECT count(*) FROM checked").fetchone()[0]
if count == 0:
raise ValueError("データ行がありません")
print("ファイル別の検算（memoのNULLは欠落列と空欄を含む）")
writer = csv.writer(sys.stdout, lineterminator="\n")
writer.writerow(["source", "rows", "memo_nulls", "total_yen"])
for path in files:
row = con.execute("""
SELECT count(*), count(*) FILTER (WHERE memo IS NULL),
coalesce(sum(amount_yen), 0)
FROM checked WHERE filename = ?
""", [str(path.resolve())]).fetchone()
writer.writerow([path.name, *row])
print("カテゴリ別の集計")
writer.writerow(["category", "rows", "total_yen"])
writer.writerows(con.execute("""
SELECT category, count(*), sum(amount_yen)
FROM checked GROUP BY category ORDER BY category
""").fetchall())
total = con.execute("SELECT sum(amount_yen) FROM checked").fetchone()[0]
print(f"全体: {count}行 / {total}円")
def main():
parser = argparse.ArgumentParser(description="月別CSVを列名で統合し、検算します")
parser.add_argument("folder", type=Path, help="入力CSVだけを置くフォルダー")
args = parser.parse_args()
try:
merge(args.folder)
except (OSError, UnicodeError, csv.Error, ValueError, duckdb.Error) as error:
print(f"集計を中止しました: {error}", file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())入力ファイル名はSQLの?へ値として渡しています。ファイル名をSQL文字列へ直接つなげません。CREATE TABLEやALTER TABLEの対象はメモリ内の一時DBで、ディスク上にDBファイルは作りません。
手順4：集計結果を確認する
作業フォルダーで実行します。
& .\.venv\Scripts\python.exe .\merge_monthly.py .\data
$LASTEXITCODE表示結果は次のとおりです。最後の0は正常終了の終了コードです。
ファイル別の検算（memoのNULLは欠落列と空欄を含む）
source,rows,memo_nulls,total_yen
2026-06.csv,2,2,1700
2026-07.csv,2,1,2500
2026-08.csv,2,0,2300
カテゴリ別の集計
category,rows,total_yen
文房具,2,1200
書籍,3,2800
食品,1,2500
全体: 6行 / 6500円
0検算は「プログラムがエラーを出さなかった」だけで終わらせません。
- ファイル別の件数が2行ずつ、全体が6行になっている。
- ファイル別の合計は1,700＋2,500＋2,300＝6,500円。
- カテゴリ別の合計も1,200＋2,800＋2,500＝6,500円。
- 6月の
memo_nullsは2、7月は1、8月は0になっている。
これらは入力の分かっている教材で照合する値です。実データでは、元システムの月次件数や確定金額など、独立した正解とも比較します。ファイルの入れ忘れは、残りのファイルだけを読んでも正常終了するため、合計値の表示だけでは検出できません。
コードの要点
欠けてよい列を限定する
requiredとallowedは、それぞれ必須列と受け付ける列の集合です。全ファイルを確認するので、ほかの月にamountがあっても、ある月だけ金額列を欠いていれば止まります。memoは任意ですべてのファイルにない場合もあり、そのときだけNULLを持つ列を追加します。
union_by_nameを外すと、異なる列数をそろえる今回の読み方にはなりません。列名での統合は入力の許容範囲を広げる指定なので、セットで「どの列の欠落を許すか」を決める必要があります。
TRY_CASTだけでは金額のルールを守れない
TRY_CASTは変換できなかった値をNULLにします。ところが、集計関数sumはNULLを無視するため、変換結果をそのまま足すと不正な入力を集計から落としてしまいます。このコードは金額の変換結果がNULLの行を見つけたら止めます。
さらに、数値変換と「円単位の整数表記かどうか」は別です。小数や桁区切りを受け付けないルールを守るため、先にregexp_full_matchで文字列全体を確認します。[+-]?[0-9]+は「省略できる符号と、1文字以上の半角数字」です。
変換先はBIGINTです。範囲は−9,223,372,036,854,775,808〜9,223,372,036,854,775,807で、範囲外も中止します。前後の空白はtrimで除きますが、全角数字、通貨記号、小数、カンマ区切りを勝手に修正する処理はありません。
重複を勝手に削除しない
GROUP BY order_id HAVING count(*) > 1で、同じ注文IDが複数回登場していないかを調べます。GROUP BYは同じ値の行をグループにし、HAVINGは集計後の条件で絞る構文です。
IDが同じなら古い行を削除する、といった推測はしません。月をまたいでIDを再利用するデータや、一つの注文に複数明細があるデータでは、この一意性ルール自体を変更する必要があります。
ファイル別の件数を残す
filename=trueで取得した入力元を残し、各ファイルのcount(*)とsum(amount_yen)を計算します。?を使った条件でそのファイルだけを選び、出力時は見やすいファイル名にしています。
FILTER (WHERE memo IS NULL)は、メモがNULLの行だけを数える指定です。coalesce(sum(...), 0)はデータ行のないファイルの合計を0と表示するために使います。不正な金額を0に置き換える用途には使っていません。
動作確認：わざと不正なデータを渡す
正常なdataを残し、テスト用フォルダーを作ります。既にdata-badがある場合は別名にしてください。
mkdir data-bad
Copy-Item .\data\*.csv .\data-bad\data-bad/2026-07.csvだけをエディターで開き、注文004の金額を700から未確定へ変更して保存します。
amount,order_id,category,memo
1800,003,書籍,追加注文
未確定,004,文房具,次を実行します。
& .\.venv\Scripts\python.exe .\merge_monthly.py .\data-bad
$LASTEXITCODE合計を出さず、次のように入力元と問題の値を表示します。
集計を中止しました: 入力値を確認してください:
2026-07.csv: order_id='004', category='文房具', amount='未確定'
2700に戻して再実行すると、正常な6行・6,500円へ戻ります。元のdataには変更を加えていないので、そちらで再実行しても構いません。
次のケースも確認しました。複数の問題を一度に混ぜず、一つずつ試すと原因を追いやすくなります。
| 入力の変更 | この教材の結果 |
|---|---|
| 金額を1.5、空欄、範囲外整数にする | 金額の確認で中止、終了コード2 |
| 金額を引用符付きの"1,200"にする | CSVとして読めても金額の確認で中止 |
| categoryを空白だけにする | 必須値の確認で中止 |
| 別ファイルにも同じorder_idを入れる | 重複として中止。自動削除しない |
| ある月のamount列を削除する | ヘッダーの確認で中止 |
| 未知の列や重複した列名を作る | ヘッダーの確認で中止 |
| 行の列数が不足・過剰、引用符が閉じていない | 読み込みで中止 |
| 全ファイルがヘッダーだけ | データ行なしとして中止 |
| 一つの月だけヘッダーのみ、別の月にはデータあり | 空の月を0行・0円として表示 |
| UTF-8 BOM、日本語ファイル名、引用されたカンマ入りカテゴリ | 入力ルール内で処理できる |
トラブル対処と制約
ModuleNotFoundError: No module named 'duckdb'なら、実行しているPythonとパッケージを入れたPythonが同じか確認します。本文のコマンドどおり、.venv内のPythonでインストールと実行をそろえてください。
「CSVがありません」なら、実行場所、指定したフォルダー、拡張子を確認します。コードはサブフォルダーを探索しません。入力フォルダーに集計結果やバックアップのCSVを置くと、それらも読まれるため、入力専用の場所にします。
文字コードのエラーが出た場合は、元データを残してUTF-8の別ファイルを作ります。CP932の日本語CSVはこの教材では受け付けません。エラー文の細部はDuckDBの版や入力内容によって変わります。
この処理は、全入力と変換結果をメモリ上のテーブルに置きます。巨大ファイルの速度・メモリ使用量は評価していません。元データを処理中に編集する運用、圧縮CSV、Excelファイル、文字コードの自動推定、異なる通貨や意味の異なる列をまとめる処理も対象外です。金額の単位やIDの意味は、読み込む前に人がそろえる必要があります。
画面出力には説明文と複数の集計が含まれるため、全体を一つのCSVファイルとして保存する設計ではありません。単一の成果物を作る場合は、必要な表を選び、入力と別の保存先へ書き出す工程を追加します。
まとめ
列構成の違う月別CSVは、DuckDBのunion_by_nameで列名をそろえて読み込めます。その後に、必須値・金額・重複を確認し、ファイル別の件数と合計を検算するところまでが今回の完成形です。
まずは自分のCSVで「必須列」「欠けてよい列」「IDの一意性」「金額の単位」を書き出してみてください。今回のルールと合う小さなコピーで試し、入力元の件数・合計と照合してから対象を増やします。
データの扱いを復習するなら読み込みとクレンジング、集計の考え方は基本統計量と集計へ進めます。単一CSVのCLIとimportのタイミングを比べたい場合は、Python 3.15のlazy importを試す記事も参照できます。
コメント