如何在C＃中高效下载，读取和处理CSV

Question

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

这是一项每7天运行一次的服务，并收集挪威公司注册中的所有公司（这里有一个漂亮的250MB，110万行CSV：http://hotell.difi.no/download/brreg/enhetsregisteret）

我的应用程序可以轻松下载文件并将其添加到List <>，并对其进行处理，但它使用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“批量更新”时，（SqlBulkTools -NuGet非常适合我的需求），每X行