c# - 如何从 LumenWorks(快速 CSV 阅读器)创建带有 IDataReader 的自定义类以在具有多个不同文件版本的同时使用 SqlBulkCopy？

Question

由于内存问题，我决定将 DataTable 替换/转换为 IDataReader。

经过一段时间的 Google 和 MSDN 搜索，我在 http://www.michaelbowersox.com/2011/12/22/using-a-custom-idatareader-to-stream-data-into-a-database/ 找到了这个和 Bulk Insert Sql Server millions of record .

因为我使用的是 LumenWorks Fast CSV Reader，所以我还没有弄清楚如何告诉 CsvReader 让 IDataReader 使用 2 个不同的字段版本。 :-( csvReader.FieldCount 是这里的关键，但我不知道如何告诉 CsvReader 使用具有 IDataReader 接口(interface)的两个新类中的任何一个。请参阅下面的原始脚本和修改后的脚本...谢谢...

//原始脚本...

var dbConnection = new SqlConnection(_dbConnectionString);

using (var dbBulkCopy = new SqlBulkCopy(dbConnection)
{
   using (CsvReader csvReader = new CsvReader(new StreamReader(filePath), false, '|', '"', '\\', '#', ValueTrimmingOptions.UnquoteOnly))
   {
       while(csvReader.ReadNextRecord())
       {
           if (csvReader.FieldCount == 48)
           {
               //Version 1...
               dataRow["DealerId"] = csvReader[0];
               dataRow["DealerName"] = csvReader[1];
               //Etc...
           }
           else if (csvReader.FieldCount == 51)
           {
               //Version 2...
               dataRow["DealerId"] = csvReader[0];
               dataRow["DealerName"] = csvReader[1];
               //Etc...
           }
           else { throw new Exception("Field Total Count Mismatched"); }

           dataTable.Rows.Add(dataRow);
       }

       dbConnection.Open();

       dbBulkCopy.WriteToServer(dataTable);
   }
}

//新脚本...

 var dbConnection = new SqlConnection(_dbConnectionString);

using (var dbBulkCopy = new SqlBulkCopy(dbConnection)
{
   dbConnection.Open();

   using (CsvReader csvReader = new CsvReader(new StreamReader(filePath), false, '|', '"', '\\', '#', ValueTrimmingOptions.UnquoteOnly))
   {
       csvReader.ReadNextRecord();

       dbBulkCopy.WriteToServer(
           if (csvReader.FieldCount == 48)
           {
               //Version 1...

               csvReader....???  //Assign a custom class having IDataTable...
           }
           else if (csvReader.FieldCount == 51)
           {
               //Version 2...
               csvReader....???  //Assign a custom class having IDataTable...
           }
           else { throw new Exception("Field Total Count Mismatched"); }
        );
   }
}

//示例脚本...

using (var file = new StreamReader(path))
using (var csv = new CsvReader(file, true)) // true = has header row
using (var bcp = new SqlBulkCopy(connection)) {
    bcp.DestinationTableName = "TableName";
    bcp.WriteToServer(csv);
}

Answer 1

因为它会有点长，所以我把它写成一个答案。

*我假设尽管您有两种具有不同字段顺序的 csv 文件，但目标表是相同。*[EDIT] 无需假设，您在评论中指出。

为了能够理解您的上下文，我从 here. 中获取了一些示例数据

假设第一种文件类型如下:

Rk,Year,Age,Tm,Lg,Id,G,W,L,W-L,Finish,
1,1978,37,Atlanta Braves,NL,,162,69,93,.426,6

和第二种类似(一些列颠倒了年龄 <-> 完成并且还有额外的字段)

Rk,Year,Finish,Tm,Lg,Id,G,W,L,W-L,Age,Test1,Test2,Test3
1,1978,Very good year,Atlanta Braves,NL,,162,69,93,.426,96,,,,

所以目标表看起来像(只有列)

Rk,Year,Finish,Tm,Lg,Id,G,W,L,W-L,Age,Test1,Test2,Test3

我在这里看到了两个选项 (末尾有 +1 个选项):

选项 1

1. 添加步骤 0，通过定义字段格式使所有输入文件在字段级别统一。这可以通过创建与数据库中相同的字段来完成。

[假设 Test4 和 Test5 是存在于目标表中但在两个 CSV 文件中都缺失的列]

Rk,Year,Finish,Tm,Lg,Id,G,W,L,W-L,Age,Test1,Test2,Test3,Test4,Test5

1. 解析您拥有的所有文件(两种类型)，按照您定义的格式将它们重写为一个(或多个，由您决定)。这样，您只有 1 个(或多个)具有唯一格式的文件。

2. 您现在可以解析此文件以使用 csv 阅读器将其插入数据库，因为字段不兼容问题已通过您以唯一格式获得的最后一个文件处理。

选项 2

您将执行两次 SqlBulkCopy 操作。第一轮您将读取具有 48 个字段的文件，下一轮您将读取具有 51 个字段的文件。

            var FilesWith48Fields = Directory.GetFiles(@"D:\Data\48Fields", "*.csv");

            foreach (var fileName in FilesWith48Fields)
            {
                using (var file = new StreamReader(fileName))
                using (var csv = new CsvReader(file, true)) // true = has header row
                using (var bcp = new SqlBulkCopy(connection, SqlBulkCopyOptions.KeepNulls))
                {
                    bcp.DestinationTableName = "fletchsodTable";
                    // map the first field of your CSV to the column in Database
                    var mapping1 = new SqlBulkCopyColumnMapping(0, "FirstColumnName");
                    bcp.ColumnMappings.Add(mapping1);

                    var mapping2 = new SqlBulkCopyColumnMapping(1, "SecondColumnName");
                    bcp.ColumnMappings.Add(mapping2);  
                    ....

                    bcp.WriteToServer(csv);
                }
            }

对包含 51 个字段的文件重复同样的操作

var FilesWith51Fields = Directory.GetFiles(@"D:\Data\51Fields", "*.csv");
......

有关 SqlBulkCopyColumnMapping can be found here 的更多信息.

选项 3

如果您想冒险创建您的数据阅读器，这里有一些链接:

Daniel Wertheim's blog

A sample implentation on codeproject

Another one

And finally MSDN

个人笔记由于时间不够，对于我的类似问题，我放弃了第三个选项，因为您必须进行所有单元测试和优化以及其他调整，这可能需要时间(至少对我来说，这是案例)

选项 4也许，使用我在 OPTION 2 中指出的列映射，您可能希望通过测试字段计数来实现您的方式。但本能地，我建议不要用硬编码整数来计算字段。