通过 EF Core 将 1.3GB CSV 文件导入 sqlite

Question

CSV 文件

我有一个大小约为 1.3 GB 的 CSV 文件：

Mode                 LastWriteTime         Length Name
----                 -------------         ------ ----
-a----         10/4/2021   1:23 PM     1397998768 XBTUSD.csv

这是 Kraken 交易所比特币交易数据的完整列表。

CSV 中的数据如下所示：

> Get-Content .\XBTUSD.csv | Select-Object -First 10
1381095255,122.00000,0.10000000
1381179030,123.61000,0.10000000
1381201115,123.91000,1.00000000
1381201115,123.90000,0.99160000
1381210004,124.19000,1.00000000
1381210004,124.18000,1.00000000
1381311039,124.01687,1.00000000
1381311093,124.01687,1.00000000
1381311094,123.84000,0.82300000
1381431835,125.85000,1.00000000

有关该文件的更多信息可在此处获得：

https://support.kraken.com/hc/en-us/articles/360047543791-Downloadable-historical-market-data-time-and-sales

文件可以从这里下载：

https://drive.google.com/drive/folders/1jI3mZvrPbInNAEaIOoMbWvFfgRDZ44TT

查看文件XBT.zip。在那个档案里面是XBTUSD.csv。

基线测试 - 直接导入 sqlite

如果我在 sqlite 中创建下表：

CREATE TABLE CsvTrades (
    "TimeStamp" TEXT NOT NULL,
    "Price"     TEXT NOT NULL,
    "Volume"    TEXT NOT NULL
);

并运行以下命令来导入 CSV（以及需要多长时间）：

$a = Get-Date

sqlite3.exe .\kraken-trades.db -cmd '.mode csv' '.import C:/Users/dharm/XBTUSD.csv CsvTrades'

$b = Get-Date

($b - $a).TotalMinutes

我得到以下信息：

1.56595191666667

1.5 分钟。还不错！

使用 EF Core

在下面的代码中，我使用的是CsvHelper 包：

https://joshclose.github.io/CsvHelper/getting-started/

这是 CSV 文件行的类：

public class CsvRow
{
    [CsvHelper.Configuration.Attributes.Index(0)]
    public long TimeStamp { get; set; }

    [CsvHelper.Configuration.Attributes.Index(1)]
    public decimal Price { get; set; }

    [CsvHelper.Configuration.Attributes.Index(2)]
    public decimal Quantity { get; set; }
}

这是Trade 实体的类：

[Index(nameof(TimeStamp))]
public class Trade
{
    public int Id { get; set; }
    public decimal Price { get; set; }
    public decimal Quantity { get; set; }
    public DateTime TimeStamp { get; set; }
}

DbContext 很简单：

public class AppContext : DbContext
{
    public DbSet<Trade> Trades { get; set; }

    protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
    {
        var folder = Environment.SpecialFolder.LocalApplicationData;

        var path = Environment.GetFolderPath(folder);

        var db_path = $"{path}{System.IO.Path.DirectorySeparatorChar}kraken-trades.db";
                
        optionsBuilder.UseSqlite($"Data Source={db_path}");
    }
}

最后是执行导入的函数：

void initialize_from_csv()
{
    var config = new CsvConfiguration(CultureInfo.InvariantCulture)
    {
        HasHeaderRecord = false
    };

    using (var reader = new StreamReader(@"C:\Users\dharm\XBTUSD.csv"))
    using (var csv = new CsvReader(reader, config))
    {
        var records = csv.GetRecords<CsvRow>().Select(row => new Trade()
        {
            Price = row.Price,
            Quantity = row.Quantity,
            TimeStamp = DateTimeOffset.FromUnixTimeSeconds(row.TimeStamp).UtcDateTime
        });

        using (var db = new AppContext())
        {
            Console.WriteLine(DateTime.Now);
                        
            while (true)
            {
                //var items = records.Take(10_000).ToList();

                var items = records.Take(100_000).ToList();

                if (items.Any() == false) break;

                Console.WriteLine("{0:yyyy-MM-dd}", items[0].TimeStamp);

                db.AddRange(items);
                db.SaveChanges();
            }

            Console.WriteLine(DateTime.Now);
        }
    }
}

问题

当我让它运行时，它确实会继续将项目添加到数据库中。但是，它很慢；我还没有计时完成，但我可以看到它需要一个多小时。

在使用 EF Core 的同时，有没有一种好方法可以加快速度？

注意事项

上面引用的代码在一个文件中可用：

https://github.com/dharmatech/kraken-trades-database/blob/003-minimal/KrakenTradesDatabase/Program.cs

这是一个 .NET 6 项目。如果您在构建和运行它时遇到任何问题，请告诉我。

时间

我添加了一些代码来计时批量添加。看起来每 100,000 条记录大约需要 7 秒。

Starting batch at 2013-10-06. Batch took 00:00:08.7689932.
Starting batch at 2015-12-08. Batch took 00:00:06.7453421.
Starting batch at 2016-04-19. Batch took 00:00:06.7833506.
Starting batch at 2016-06-25. Batch took 00:00:06.7083806.
Starting batch at 2016-08-22. Batch took 00:00:06.7826717.
Starting batch at 2016-11-20. Batch took 00:00:06.4212123.

wc 表示有 41,695,261 行：

$ wc -l XBTUSD.csv
41695261 XBTUSD.csv

所以按照这个速度，大约需要 48 分钟。

为什么选择 EF Core？

有些人问，为什么要为此使用 EF Core？为什么不直接导入？

特意简化了上面的示例，重点关注导入速度。

我有更详细的版本，其中存在与其他实体的关系。在这种情况下：

• 使用 EF Core 设置其他表和外键属性更加简单。

• 我可以更轻松地在数据库后端（SQL Server、PostgreSQL、sqlite）之间切换。

参见这个分支，其中导入了多个符号。 Trade 和 Symbol 之间存在关系。也可能存在其他关系。

https://github.com/dharmatech/kraken-trades-database/blob/006/KrakenTradesDatabase/Program.cs

Answer 1

EFCore.BulkExtensions

使用以下内容：

https://github.com/borisdj/EFCore.BulkExtensions

然后改变这一行：

db.AddRange(items);

到：

db.BulkInsert(items);

使导入时间从 48 分钟缩短到 5.7 分钟。

此版本的项目可在此处获得：

https://github.com/dharmatech/kraken-trades-database/blob/004-bulk-extensions/KrakenTradesDatabase/Program.cs

谢谢

感谢 Caius Jard 在上面的评论中建议 EFCore.BulkExtensions。

Answer 2

这是一个完整的 C# (10.0) 程序，它比 sqlite3 工具更快地插入 CSV 数据。它使用我的 Sylvan.Data.Csv 库，这是 .NET 上最快的 CSV 解析器。

在我的机器上，sqlite3 会在1:07.6 中插入数据，而我的代码会在1:02.9 中插入。

虽然这不符合您“仍在使用 EFCore”的要求，但我认为性能差异不言而喻。

包：

<PackageReference Include="Sylvan.Data.Csv" Version="1.1.9" />
<PackageReference Include="System.Data.SQLite" Version="1.0.115.5" />

代码：

using System.Collections.ObjectModel;
using System.Data.Common;
using System.Data.SQLite;
using System.Diagnostics;
using Sylvan.Data.Csv;

var sw = Stopwatch.StartNew();

var conn = new SQLiteConnection("Data Source=test.db");
conn.Open();

var data = CsvDataReader.Create("xbtusd.csv", new CsvDataReaderOptions { HasHeaders = false });


// create the target table
{
    using var cmd = conn.CreateCommand();
    var tbl = "create table CsvTrades (TimeStamp TEXT NOT NULL, Price TEXT NOT NULL, Volume TEXT NOT NULL)";
    cmd.CommandText = tbl;
    cmd.ExecuteNonQuery();
}

// get the schema for the target table.
ReadOnlyCollection<DbColumn> ss;
{
    using var cmd = conn.CreateCommand();
    cmd.CommandText = "select * from CsvTrades limit 0;";
    var r = cmd.ExecuteReader();
    ss = r.GetColumnSchema();
}

// create the parameterized insert command
var cmdW = new StringWriter();
cmdW.Write("insert into CsvTrades values(");
int i = 0;
foreach (var c in ss)
{
    if (i > 0)
        cmdW.Write(",");
    cmdW.Write("$p" + i++);
}

cmdW.Write(");");
var cmdt = cmdW.ToString();

// insert CSV data.
using (var tx = conn.BeginTransaction())
{
    var cmd = conn.CreateCommand();
    cmd.CommandText = cmdt;
    for (i = 0; i < data.FieldCount; i++)
    {
        var p = cmd.CreateParameter();
        p.ParameterName = "$p" + i;
        cmd.Parameters.Add(p);
    }
    cmd.Prepare();
    while (data.Read())
    {
        for (i = 0; i < data.FieldCount; i++)
        {
            cmd.Parameters[i].Value = data.GetValue(i);
        }
        cmd.ExecuteNonQuery();
    }

    tx.Commit();
}

sw.Stop();
Console.WriteLine($"Inserted {data.RowNumber} records in {sw.Elapsed}");

更新： 我意识到我的代码并没有做它需要做的一切：它没有将 unix 秒转换为日期时间。修改插入循环如下：

    while (data.Read())
    {
        cmd.Parameters[0].Value = DateTime.UnixEpoch.AddSeconds(data.GetInt64(0));
        cmd.Parameters[1].Value = data.GetString(1);
        cmd.Parameters[2].Value = data.GetString(2);
        cmd.ExecuteNonQuery();
    }

这会将速度减慢到 1:17.5，比 sqlite3 稍微慢一些，但随后 sqlite3 插入不会进行数据转换，因此您最终会得到原始整数（长整数）值。