如何在 C# 中高效下载、读取和处理 CSV

Question

我正在开发一项服务，该服务将从在线资源中收集大型 CSV 文件，然后在下载时读取这些行（最好是分批），然后将它们发送到数据库。这在任何时候都不应使用超过 256MB 的 RAM，并且不应将文件保存到磁盘。

这是一项每 7 天运行一次的服务，并收集挪威公司登记册中的所有公司（一个漂亮的 250MB 110 万行 CSV 文件可在此处找到：http://hotell.difi.no/download/brreg/enhetsregisteret）

我的应用程序可以轻松下载文件并将其添加到列表，并对其进行处理，但它使用 3.3 GB 的 RAM

public async Task<bool> CollectAndUpdateNorwegianCompanyRegistry()
{
    var request = await _httpClient.GetAsync(_options.Value.Urls["BrregCsv"]);

    request.EnsureSuccessStatusCode();

    using (var stream = await request.Content.ReadAsStreamAsync())
    using (var streamReader = new StreamReader(stream))
    {
        while (!streamReader.EndOfStream)
        {
            using (var csv = new CsvReader(streamReader)) // CsvReader is from the CsvHelper -nuget
            {
                csv.Configuration.Delimiter = ";";
                csv.Configuration.BadDataFound = null;
                csv.Configuration.RegisterClassMap<NorwegianCompanyClassMap>();

                await _sqlRepository.UpdateNorwegianCompaniesTable(csv.GetRecords<NorwegianCompany>().ToList());
            }
        }
    }

    return true;
}

关于 SqlRepository 的小说明：我已经用一个简单的“破坏者”方法替换了它，它只是清除数据，以便在调试时不使用任何额外的资源

我希望垃圾收集器会“销毁”处理文件行时使用的资源，但事实并非如此。

简单地说，我希望发生以下情况： 当 CSV 下载时，它会读取几行，然后将它们发送到一个方法，然后刷新内存中的行

我在处理大型数据集方面确实缺乏经验，所以我正在处理其他人的工作，并没有得到我期望的结果

感谢您的时间和帮助

Answer 1

所以从 Sami Kuhmonen (@sami-kuhmonen) 那里得到了一些建议，这就是我的想法：

public async Task<bool> CollectAndUpdateNorwegianCompanyRegistry()
{
    using (var stream = await _httpClient.GetStreamAsync(_options.Value.Urls["BrregCsv"]))
    using (var streamReader = new StreamReader(stream))
    using (var csv = new CsvReader(streamReader))
    {
        csv.Configuration.Delimiter = ";";
        csv.Configuration.BadDataFound = null;
        csv.Configuration.RegisterClassMap<NorwegianCompanyClassMap>();

        await _sqlRepository.UpdateNorwegianCompaniesTable(csv.GetRecords<NorwegianCompany>());
    }

    return true;
}

它会在 20 秒内下载整个文件并将其发送到 SqlRepository，不会超过 15% 的 CPU 或 30MB RAM

现在，我的下一个挑战是 SqlRepository，但是这个问题已经解决了

Answer 2

我现在正在实施的另一个解决方案，它的资源使用更可预测：

public async Task<bool> CollectAndUpdateNorwegianCompanyRegistryAlternate()
{
    using (var stream = await _httpClient.GetStreamAsync(_options.Value.Urls["BrregCsv"]))
    using (var reader = new StreamReader(stream))
    using (var csv = new CsvReader(reader))
    {
        csv.Configuration.RegisterClassMap<NorwegianCompanyClassMap>();
        csv.Configuration.Delimiter = ";";
        csv.Configuration.BadDataFound = null;

        var tempList = new List<NorwegianCompany>();

        while (csv.Read())
        {
            tempList.Add(csv.GetRecord<NorwegianCompany>());

            if (tempList.Count() > 50000)
            {
                await Task.Factory.StartNew(() => _sqlRepository.UpdateNorwegianCompaniesTable(tempList));

                tempList.Clear();
            }
        }
    }
    return true;
}

现在它使用 3 分钟，但从未达到 200MB 的峰值，并且使用 7-12% 的 CPU，即使在执行 SQL“批量更新”时，（SqlBulkTool -NuGet 非常适合我的需求），每 X 行