【ITニュース解説】Catching Column Drift in Recurring CSV Files Before an Automated Run

日々、システムを運用していると、様々なデータソースから情報を取り込み、自動で処理する機会がたくさんある。中でも、CSVファイルは手軽で広く使われているデータ形式だ。しかし、このCSVファイルを毎日決まった処理で読み込んでいると、ある日突然、誰も気づかないうちに処理が壊れてしまうことがある。

例えば、CSVファイルのヘッダー、つまり列の名前が「所有者名」から「所有者」に変わったり、途中に新しい列が追加されたり、元々あった2つの列の順番が入れ替わったりすることがある。このような「カラムずれ」が起こると、自動処理は何も知らずに間違った列のデータを読み込み、その結果、間違った情報がシステムに取り込まれてしまう。そして、その間違いは、データがおかしな結果として現れるまで、誰も気づかないという静かな形で進行する。これは、自動化を進める上で非常に危険な状況だ。

このような問題を未然に防ぎ、安全にデータ処理を行うための方法が、今回紹介するPythonを使ったパターンだ。このパターンでは、主に以下の4つのステップで問題を解決する。

1. CSVファイルの構造（スキーマ、つまりヘッダー情報）を記憶する。
2. 次回以降に読み込むファイルの構造が記憶したものと異なっていたら、それを検知し、具体的に何が変わったかを明確に伝える。
3. 過去に処理したことのない新しい行や、内容が変更された行だけを処理対象とする。
4. 実際にデータを動かす前に、処理内容の「試運転（ドライラン）」を行う。

では、それぞれのステップを詳しく見ていこう。

まず、1. スキーマフィンガープリントの保存 についてだ。 新しいCSVファイルを初めてシステムに取り込む際、そのファイルのヘッダー（列名）の並び順と、そのヘッダーから生成した「指紋（フィンガープリント）」を保存しておく。フィンガープリントとは、ファイルの内容を元に作られる一意の短い文字列で、これが同じならファイルの内容も同じと見なせるものだ。 このとき、ヘッダーの名前はすべて小文字にし、前後の余分な空白も取り除く「正規化」という処理を行う。これにより、「Owner Name」と「owner name」のように大文字・小文字が違うだけや、余分な空白があるだけの見た目の変更では、新しいスキーマだと誤解するのを防ぐことができる。正規化されたヘッダーのリストと、そこから作ったフィンガープリントはJSONファイルに保存され、これが「このCSVファイルはこの形式で受け入れられました」という証拠になる。

次に、2. カラムずれの検知と具体的な報告 だ。 新しいCSVファイルを処理する前に、保存しておいたスキーマ情報と比較する。もし以前にスキーマが保存されていない場合は、システムは処理を停止し、「まだ承認されたスキーマがありません。ヘッダーを確認して、保存してください」と人間への対応を促す。これは、最初のファイルを自動的に信用してしまうことを避けるためだ。 比較の結果、現在のファイルのフィンガープリントが保存されたものと同じで、かつ重複するヘッダーもなければ、変更はないと判断する。しかし、もしフィンガープリントが異なっていたり、重複するヘッダーが見つかったりした場合は、システムはただ「スキーマが変わりました」と漠然と伝えるのではなく、「どの列が追加されたか」「どの列が削除されたか」「どの列が移動したか」「重複している列はどれか」といった具体的な変更内容をリストアップして報告する。 そして、この変更が検知されたら、システムは自動的な列のマッピング変更を試みるのではなく、そこで処理を停止する。これは、システムが間違った判断をしてデータを壊すことを防ぐためだ。人間が変更内容を確認し、適切な対応（例えば、マッピングの更新）を行った上で、新しいスキーマとして承認するまで、次の処理には進まない。

続いて、3. 新規または変更された行のみを処理 する方法だ。 CSVファイルには、それぞれ固有の情報を識別するための「ID」となる列があるはずだ。例えば、顧客IDや案件番号などがそれにあたる。このID列を使って、以前に処理した行か、それとも新しい行か、あるいは内容が変わった行かを判断する。 具体的には、各行のすべてのデータ内容（空白をトリミングし、列の順番は考慮せずソートして）からハッシュ値を計算する。このハッシュ値も一種のフィンガープリントで、行の内容が少しでも変わればハッシュ値も変わる。このIDとそのハッシュ値を「以前に処理した行」としてSQLiteなどのデータベースに保存しておく。 新しいCSVファイルを読み込む際、各行について以下の判断を行う。

• もしその行にIDがなければ、「IDなし」として報告し、処理対象から外す。
• もし同じファイル内に同じIDを持つ別の行があれば、「重複」として報告し、二重に処理しない。
• データベースに同じIDの行がすでに記録されており、そのときのハッシュ値と現在の行のハッシュ値が同じであれば、その行は「変更なし」と判断し、スキップする。
• データベースに同じIDの行が記録されているが、ハッシュ値が異なる場合は、その行は「変更された」と判断し、処理対象とする。
• データベースに同じIDの行が全く記録されていなければ、その行は「新規」と判断し、処理対象とする。 この方法により、毎日の処理で本当に必要な新規データや更新データだけを効率的に処理できる。

さらに、4. ドライラン（試運転）の実施 は非常に重要だ。 実際のデータ連携やデータベースへの書き込みなど、本番の処理を行う前に、まずは「試運転」として何が起こるかを事前に確認する。この試運転では、データの読み込みは行うが、データベースへの書き込みや外部サービスへの連携は一切行わない。もし以前のデータ履歴を保存しているデータベースがまだ存在しない場合でも、エラーとせずにすべての行を「新規」として扱うが、実際にデータベースを作成したり書き込んだりすることはしない。 試運転の結果として、実際に処理される予定の行数、処理上限を超えて待機する行数、変更がなくスキップされる行数、IDがないために保留される行数、重複しているために保留される行数などがレポートされる。これにより、システムが意図した通りに動作するかを事前に検証でき、問題があれば本番処理の前に修正できる。システムの運用設定を変更した場合には、必ず再度試運転を行うべきだ。

最後に、5. ファイルを日付順に処理し、すべてを記録する という点も重要だ。 もし複数のCSVファイルが特定のフォルダに蓄積されていくような状況であれば、ファイル名に含まれる日付情報（例えば「2023-10-26.csv」のようなISO形式）を使って、古いファイルから新しいファイルの順に処理するべきだ。これにより、もし同じIDの行が異なるファイルで更新されていた場合、最も新しいファイルの内容が正しく適用される。また、ファイルがコピーされたり再ダウンロードされたりして更新日時が変わるような状況でも、ファイル名の日付が基準となるため、処理順序が狂うことがない。 実際の処理では、各行の処理が成功した後にのみ、そのIDとハッシュ値をデータベースに記録する。そして、1つのCSVファイルの処理が終わるたびに、データベースへの変更を確定（コミット）する。こうすることで、もし処理の途中でシステムが停止しても、確定されていない未処理の行だけが次の実行で再試行されるようになり、データが失われたり二重に処理されたりするリスクを減らせる。さらに、どの行がどのように処理されたか（スキップされたのか、更新されたのか、新規として処理されたのかなど）を詳細にログとして残しておくことは、後になって「なぜこの行はスキップされたのか」といった疑問に答えるために不可欠な情報となる。

これらのルールを守ることで、自動処理はより堅牢になり、予期せぬデータの破損を防ぐことができる。よくある間違いとして、列名をあいまいな基準で自動的に再マッピングしようとしたり、行番号を行のIDとして使ったり、設定変更後に試運転を省略したり、空白の正規化を怠ってハッシュ値を計算したりすることが挙げられるが、これらはすべて深刻な問題を引き起こす可能性があるため注意が必要だ。

どんなに優れた自動化ツールを使っても、根本的なルールは変わらない。それは「入力データの形が変わったら、そこで処理を停止し、何が変わったかを具体的に報告し、最終的な判断は人間に委ねる」という原則だ。この原則を守ることが、安定したシステム運用の鍵となる。