[C#] 大量データを効率よく投入する

前回は、追加、更新、削除、ストアドプロシージャ、トランザクションを扱いました。

今回は、大量データを効率よく投入する方法を見ていきます。

数件の INSERT であれば、通常の SqlCommand で十分です。
しかし、数万件、数十万件のデータを 1 行ずつ INSERT すると、通信回数やログ処理のコストが大きくなります。

SQL Server では、SqlBulkCopy を使うことで、大量データを効率よく投入できます。

一括投入が必要になる場面

一括投入は、次のような場面で使われます。

CSV からデータを取り込む
外部システムから受け取った大量データを登録する
ログや計測値をまとめて保存する
テストデータを短時間で作成する
旧システムから移行する

1 件ずつ INSERT するよりも、まとめて送る方が速くなることが多いです。

投入先テーブルを用意する

サンプル用の投入先テーブルを作ります。

CREATE TABLE InventoryBulk
(
    Id INT IDENTITY(1,1) NOT NULL PRIMARY KEY,
    Make NVARCHAR(50) NOT NULL,
    Color NVARCHAR(50) NOT NULL,
    PetName NVARCHAR(50) NOT NULL
);
GO

既存の Inventory テーブルへ直接投入してもよいですが、学習用には別テーブルにすると試しやすくなります。

DataTable から一括投入する

もっとも分かりやすい例は、DataTable を使う方法です。

using System.Data;
using Microsoft.Data.SqlClient;

static DataTable CreateCarTable()
{
    DataTable table = new DataTable();

    table.Columns.Add("Make", typeof(string));
    table.Columns.Add("Color", typeof(string));
    table.Columns.Add("PetName", typeof(string));

    table.Rows.Add("Honda", "Blue", "Bulk Vega");
    table.Rows.Add("Ford", "Red", "Bulk Rusty");
    table.Rows.Add("Toyota", "Black", "Bulk Night");

    return table;
}

投入処理です。

static async Task BulkInsertWithDataTableAsync(string connectionString)
{
    DataTable table = CreateCarTable();

    await using SqlConnection connection = new SqlConnection(connectionString);
    await connection.OpenAsync();

    using SqlBulkCopy bulkCopy = new SqlBulkCopy(connection);

    bulkCopy.DestinationTableName = "InventoryBulk";

    bulkCopy.ColumnMappings.Add("Make", "Make");
    bulkCopy.ColumnMappings.Add("Color", "Color");
    bulkCopy.ColumnMappings.Add("PetName", "PetName");

    await bulkCopy.WriteToServerAsync(table);
}

ColumnMappings は、入力側の列と投入先テーブルの列を対応付けます。

一括投入を確認する

SELECT *
FROM InventoryBulk;

データが投入されていれば成功です。

件数だけ確認するなら、C# から ExecuteScalarAsync() でも確認できます。

await using SqlCommand command = connection.CreateCommand();
command.CommandText = "SELECT COUNT(*) FROM InventoryBulk";

int count = (int)await command.ExecuteScalarAsync();

Console.WriteLine(count);

DataTable の利点と欠点

DataTable は分かりやすく、少量から中規模の一括投入では便利です。

利点:

コードが理解しやすい
列定義を明示できる
SqlBulkCopy と相性が良い

欠点:

全データをメモリ上に持つ
大量データではメモリ使用量が増える
型付きモデルとの相性はやや古典的

非常に大きなデータでは、次に見る IDataReader を使う方法も検討します。

IDataReader から一括投入する考え方

SqlBulkCopy は、IDataReader からもデータを読み取れます。

IDataReader を自作すると、データを 1 行ずつ生成しながら投入できます。
これにより、全データをメモリ上に持たずに済みます。

ここでは簡略化したカスタム読み取りクラスを作ります。

using System.Data;

public sealed class CarDataReader : IDataReader
{
    private readonly IEnumerator<Car> enumerator;

    public CarDataReader(IEnumerable<Car> cars)
    {
        enumerator = cars.GetEnumerator();
    }

    public bool Read() => enumerator.MoveNext();

    public int FieldCount => 3;

    public object GetValue(int i)
    {
        return i switch
        {
            0 => enumerator.Current.Make,
            1 => enumerator.Current.Color,
            2 => enumerator.Current.PetName,
            _ => throw new IndexOutOfRangeException()
        };
    }

    public string GetName(int i)
    {
        return i switch
        {
            0 => "Make",
            1 => "Color",
            2 => "PetName",
            _ => throw new IndexOutOfRangeException()
        };
    }

    public void Dispose() => enumerator.Dispose();
    public void Close() => Dispose();
    public bool IsClosed => false;
    public int RecordsAffected => -1;
    public int Depth => 0;
    public bool NextResult() => false;
    public DataTable? GetSchemaTable() => null;

    public object this[int i] => GetValue(i);
    public object this[string name] => GetValue(GetOrdinal(name));
    public int GetOrdinal(string name) => name switch
    {
        "Make" => 0,
        "Color" => 1,
        "PetName" => 2,
        _ => throw new IndexOutOfRangeException()
    };

    public string GetDataTypeName(int i) => GetFieldType(i).Name;
    public Type GetFieldType(int i) => typeof(string);
    public int GetValues(object[] values)
    {
        int count = Math.Min(values.Length, FieldCount);
        for (int i = 0; i < count; i++)
        {
            values[i] = GetValue(i);
        }
        return count;
    }

    public bool IsDBNull(int i) => false;

    public bool GetBoolean(int i) => (bool)GetValue(i);
    public byte GetByte(int i) => (byte)GetValue(i);
    public long GetBytes(int i, long fieldOffset, byte[]? buffer, int bufferoffset, int length) => 0;
    public char GetChar(int i) => (char)GetValue(i);
    public long GetChars(int i, long fieldoffset, char[]? buffer, int bufferoffset, int length) => 0;
    public Guid GetGuid(int i) => (Guid)GetValue(i);
    public short GetInt16(int i) => (short)GetValue(i);
    public int GetInt32(int i) => (int)GetValue(i);
    public long GetInt64(int i) => (long)GetValue(i);
    public float GetFloat(int i) => (float)GetValue(i);
    public double GetDouble(int i) => (double)GetValue(i);
    public string GetString(int i) => (string)GetValue(i);
    public decimal GetDecimal(int i) => (decimal)GetValue(i);
    public DateTime GetDateTime(int i) => (DateTime)GetValue(i);
    public IDataReader GetData(int i) => throw new NotSupportedException();
}

実務では、このような実装を毎回手書きするより、既存ライブラリや単純な DataTable で十分なことも多いです。
ここでは、SqlBulkCopy が「行を読み取れるもの」から投入できる、という考え方を押さえるための例です。

IDataReader を使って投入する

static IEnumerable<Car> GenerateCars()
{
    yield return new Car { Make = "Mazda", Color = "White", PetName = "Snow" };
    yield return new Car { Make = "Nissan", Color = "Green", PetName = "Leafy" };
    yield return new Car { Make = "Subaru", Color = "Blue", PetName = "Sky" };
}

投入処理です。

static async Task BulkInsertWithReaderAsync(string connectionString)
{
    await using SqlConnection connection = new SqlConnection(connectionString);
    await connection.OpenAsync();

    using SqlBulkCopy bulkCopy = new SqlBulkCopy(connection);

    bulkCopy.DestinationTableName = "InventoryBulk";
    bulkCopy.ColumnMappings.Add("Make", "Make");
    bulkCopy.ColumnMappings.Add("Color", "Color");
    bulkCopy.ColumnMappings.Add("PetName", "PetName");

    using CarDataReader reader = new CarDataReader(GenerateCars());

    await bulkCopy.WriteToServerAsync(reader);
}

データを逐次生成できるため、非常に大きなデータを扱う場合に有利です。

一括投入とトランザクション

一括投入もトランザクションで囲めます。

await using SqlConnection connection = new SqlConnection(connectionString);
await connection.OpenAsync();

await using SqlTransaction transaction =
    (SqlTransaction)await connection.BeginTransactionAsync();

try
{
    using SqlBulkCopy bulkCopy = new SqlBulkCopy(
        connection,
        SqlBulkCopyOptions.Default,
        transaction);

    bulkCopy.DestinationTableName = "InventoryBulk";
    bulkCopy.ColumnMappings.Add("Make", "Make");
    bulkCopy.ColumnMappings.Add("Color", "Color");
    bulkCopy.ColumnMappings.Add("PetName", "PetName");

    await bulkCopy.WriteToServerAsync(CreateCarTable());

    await transaction.CommitAsync();
}
catch
{
    await transaction.RollbackAsync();
    throw;
}

途中で失敗した場合に全体を取り消したいなら、トランザクションを使います。

一括投入時の注意点

大量データを扱うときは、次の点に注意します。

投入先テーブルの列と型を合わせる
必要なら列対応を明示する
メモリ使用量を意識する
失敗時のロールバック方針を決める
制約やインデックスの影響を考える
重複データをどう扱うか決める

一括投入は速い反面、失敗時の影響も大きくなります。
事前検証、ログ、リトライ、エラー行の扱いを設計しておくと運用しやすくなります。

まとめ

大量データを 1 件ずつ INSERT するのは効率が悪くなりがちです。
SQL Server では SqlBulkCopy を使うことで、まとまったデータを高速に投入できます。

小規模なら DataTable が分かりやすく、大規模なら IDataReader による逐次読み取りも選択肢になります。
ADO.NET の基本を知っていると、ORM だけでは扱いにくい大量投入や特殊なデータ処理にも対応しやすくなります。