C#でNpgsqlを使ってPostgreSQLへ大量のテキストデータを高速に処理する

前回のバイナリコピーに関連したテキストコピーです。

主にCSVファイルのデータをINSERTしたり、データベースのデータをCSVファイルに出力する場合に向いている手法です。

• もくじ
  • テキストデータの書き込み
    • 変数データの書き込み
    • csvファイルを1行のづつ書き込み
    • csvファイルをまとめて書き込み
  • テキストデータの読み込み
    • 変数データへ読み込み
    • 1行のづつ読み込んでcsvファイルへ書き込み
    • まとめて読み込んでcsvファイルへ書き込み

Npgsqlの本家情報は

COPYに関する解説は

下記CPOYの文中にあるリンクはPostgreSQLのCOPYコマンドのリンクなのでおそらくPostgreSQLのCOPYコマンドを具現化していると思われます。

また、この手法はバイナリコピーよりパフォーマンスが落ちる事とCSVファイルとデータテーブルの整合性はきちんと行う事が注意点として書かれています。

私の作成するサンプルソースファイルは

基本的なテーブルは下記構成となります。

テーブル名	型	概要
id	serial	自動的にセットされる通し番号
time	timestamp	トランザクション開始時刻または入力された日付
name	text	任意の文字列
numeric	integer	任意の数値

時間はCREATE TABLEで「time timestamp DEFAULT clock_timestamp()」としており、意図的に時間をセットできますが、何もINSERTしなければ自動的に現在の時間がセットされます。

今回のサンプルでcsvファイルへアクセスする場合、どこかで

Encoding.RegisterProvider(CodePagesEncodingProvider.Instance);

の記述がないと例外が発生するので注意が必要です。

テキストデータの書き込み

変数データの書き込み

最初の検証は変数の値を書き込んでみます。

using NpgsqlConnection con = new("Server=127.0.0.1; Port=5432; User Id=test_user; Password=pass; Database=db_PostgreTest; SearchPath=public");
con.Open();
using TextWriter writer = con.BeginTextImport("COPY data (name, numeric) FROM STDIN WITH csv");
writer.NewLine = "\n";
// 変数を書き込む
for (int i = 0; i < 100000; i++)
{
    writer.WriteLine($"name{i},{i}");
}

上記サンプルコードの3行目にあるNpgsqlConnectionのBeginTextImportメソッドの戻り値TextWriterを用いてデータをセットしてデータベースに書き込みを行います。

また、同行の引数文字列にも注意が必要です。

引数文字列の最後に「WITH csv」と記述していますが、これがあるとデータは「カンマ区切り」となりますが、記述がないと「タブ区切り」にする必要があります。

4行目にはWriteLineメソッドの改行コードを指定しています。

BeginTextImportでは「\n」が1行の最後を示します。

しかしTextWriter（Windows）のデフォルトは「\r\n」なのできちんと指定しないと「\r」も送ってしまいます。

実験した所、最後のカラムが数値の場合は無視されるようですが、意図しないデータが書き込まれる可能性もあるので明確にした方がいいです。

今回の例だとWriteLineメソッドではなくWriteメソッドを使い文字列に「\n」を追加しても同じになりますのでやりやすい方法で良いと思います。

この手法で実際にデータベースに書き込まれるタイミングは「using TextWriter」を抜けた時になります。

10万個のデータを書き込んだ時間は最短で0.51秒、10回平均で0.53秒とバイナリコピーの平均0.33秒よりは遅いですがクエリ文を使ったINSERTよりは断然早いです。

csvファイルを1行のづつ書き込み

次にcsvファイルを1行づつ追加してみます。

using NpgsqlConnection con = new("Server=127.0.0.1; Port=5432; User Id=test_user; Password=pass; Database=db_PostgreTest; SearchPath=public");
con.Open();
using TextWriter writer = con.BeginTextImport("COPY data (name, numeric) FROM STDIN WITH csv");
writer.NewLine = "\n";
// csvファイルを1行づつ読み込みならが書き込む
using StreamReader reader = new("test.csv", Encoding.GetEncoding("Shift_JIS"));
string str;
while ((str = reader.ReadLine()) != null)
{
    writer.WriteLine(str.Replace("\r\n", ""));
}

4行目までは変数データの書き込みと同じです。

6行目でcsvファイルを開く準備を行います。

8行目ではReadLineソッドがnullになるまで処理を繰り返すwhile文になっています。

そして10行目で改行コードを消しつつWriteLineで書き込むデータに加えてます。

ここは「Replace("\n", "")」としてWriteLineではなくWriteメソッドでも結果は同じです。

大事なのは不要な制御コードは消し、必要な制御コードは残す事です。

10万個のデータを書き込んだ時間は最短で0.51秒、10回平均で0.54秒とSSDの恩恵か行ごとに読んで書いてを行っている割には健闘しているかと思います。

csvファイルをまとめて書き込み

PostgreSQL側が「\n」を行の区切りと識別しているので以下の記述が一番シンプルになります。

using NpgsqlConnection con = new("Server=127.0.0.1; Port=5432; User Id=test_user; Password=pass; Database=db_PostgreTest; SearchPath=public");
con.Open();
using TextWriter writer = con.BeginTextImport("COPY data (name, numeric) FROM STDIN WITH csv");
// csvファイルをまとめて読み込んで書き込む
using StreamReader reader = new("test.csv", Encoding.GetEncoding("Shift_JIS"));
writer.Write(reader.ReadToEnd().Replace("\r", ""));

データはReadToEndメソッドで読みつつ不要な「\r」を削除してWriteメソッドの引数としています。

10万個のデータを書き込んだ時間は最短で0.53秒、10回平均で0.55秒と1行づつよりは早いかと思ったのですが何度やっても微妙に負けました。

まどめて大量なReplaceを行っていますのでこれが影響しているかもしれません。

遅いとは言え10万個のデータでわずか0.01秒でしかないのでシンプルさを考えたらこちらに軍配があがるかもしれません。

また、アクセスの遅いHDDだと逆転する可能性もありますのでダメとは言い切れません。

テキストデータの読み込み

変数データへ読み込み

最初の検証はデータベースから取得したデータを変数に書き込んでみます。

using NpgsqlConnection con = new("Server=127.0.0.1; Port=5432; User Id=test_user; Password=pass; Database=db_PostgreTest; SearchPath=public");
con.Open();
using TextReader reader = con.BeginTextExport("COPY data(name, numeric) TO STDOUT WITH csv");
List<string> selectData = new();
string str;
while ((str = reader.ReadLine()) != null)
{
	selectData.Add(str);
}

上記サンプルコードの3行目にあるNpgsqlConnectionのBeginTextExportメソッドの戻り値TextReaderを用いてデータをセットしてデータベースに読み込みを行います。

書き込みと同様、引数文字列の最後に「WITH csv」と記述して取得したデータは「カンマ区切り」としますが、記述がないと「タブ区切り」となります。

実験は1行づつデータを取得し特に加工もせずstring型のListへ追加しています。

今回は同一PC内での処理ですが、ネットワークを介したデータ取得の場合はReadToEndメソッドで一気に取得してプログラム内で分解する手法もアリかもしれません。

10万個のデータを取得した時間は最短で0.031秒、10回平均で0.042秒とバイナリコピーの平均0.039秒よりは遅いですが無条件のSELECTより0.01秒早いです。

早いとは言えわずか0.01秒ですから誤差範囲とも言えるレベルです。

1行のづつ読み込んでcsvファイルへ書き込み

データベースからはReadLineメソッドで1行づつ取得し、WriteLineメソッドで1行つづcsvファイルに書き込みます。

using NpgsqlConnection con = new("Server=127.0.0.1; Port=5432; User Id=test_user; Password=pass; Database=db_PostgreTest; SearchPath=public");
con.Open();
using TextReader reader = con.BeginTextExport("COPY data(name, numeric) TO STDOUT WITH csv");
string str;
while ((str = reader.ReadLine()) != null)
{
    writer.WriteLine(str);
}

データの取得は.ReadLineメソッドの戻り値がnullになるまで行います。

10万個のデータを取得した時間は最短で0.047秒、10回平均で0.052秒とファイルアクセスが発生しているだけ変数に入れるよりは遅くなっています。

またループ処理の部分は

foreach (var s in reader.ReadToEnd().Split("\n"))
{
	writer.WriteLine(s);
}

ともう少しスマートに記述できますが、foreachでReadToEndメソッドを使用すると一番最後に「何もない文字列」を取得します。

なので、取得した文字列にちゃんと文字が入っているか確認する必要があるので、どっちがいいかの判断は難しいです。

まとめて読み込んでcsvファイルへ書き込み

読み込みと同様に短いコードでデータを取得する事ができます。

using NpgsqlConnection con = new("Server=127.0.0.1; Port=5432; User Id=test_user; Password=pass; Database=db_PostgreTest; SearchPath=public");
con.Open();
using TextReader reader = con.BeginTextExport("COPY data(name, numeric) TO STDOUT WITH csv");
using StreamWriter writer = new("test.csv", false, Encoding.GetEncoding("Shift_JIS"));
writer.Write(reader.ReadToEnd().Replace("\n", "\r\n"));

必要なデータはReadToEndメソッドでまとめて取得しつつWindows環境なので改行コードをまとめて変更して一気にcsvファイルへ書き込みます。

10万個のデータを取得した時間は最短で0.055秒、10回平均で0.062秒と1行づつ取得するより遅くなりました。

これもまとめて書き込み同様Replaceメソッドで大量に文字列変換が影響しているかもしれません。