使用 C# 将大型 csv 数据文件上传到 SQL 服务器的最佳方法是什么？该文件包含大约 30,000 行和 25 列。
【问题讨论】：
标签： c# .net sql-server csv
第一关，你不需要编程的东西。您可以使用 SQL 管理工具将 CSV 文件直接上传到 SQL 数据库。但是，如果您真的需要通过编程来实现，请阅读下文。
就个人而言，我认为这种方法是通过编程实现的最有效和最简单的方法。
一般来说，分两步即可实现
第 1 步是读取 CSV 文件并将记录保存为 DataTable。
第 2 步是将检索到的 DataTable 存储到 SQL 数据库中表作为批量条目
这是一个将 CSV 文件数据作为DataTable 返回的函数。调用它并将其保存在内存中，您可以对它做任何您想做的事情。
此函数会将读取的 CSV 文件返回到 DataTable 中。
private static DataTable GetDataTabletFromCSVFile(string csv_file_path)
{
DataTable csvData = new DataTable();
try
{
using(TextFieldParser csvReader = new TextFieldParser(csv_file_path))
{
csvReader.SetDelimiters(new string[] { "," });
csvReader.HasFieldsEnclosedInQuotes = true;
string[] colFields = csvReader.ReadFields();
foreach (string column in colFields)
{
DataColumn datecolumn = new DataColumn(column);
datecolumn.AllowDBNull = true;
csvData.Columns.Add(datecolumn);
}
while (!csvReader.EndOfData)
{
string[] fieldData = csvReader.ReadFields();
//Making empty value as null
for (int i = 0; i < fieldData.Length; i++)
{
if (fieldData[i] == "")
{
fieldData[i] = null;
}
}
csvData.Rows.Add(fieldData);
}
}
}
catch (Exception ex)
{
return null;
}
return csvData;
}
}
SQLBulkCopy - 使用此函数将检索到的数据表插入到 Sql 表中
static void InsertDataIntoSQLServerUsingSQLBulkCopy(DataTable csvFileData)
{
using(SqlConnection dbConnection = new SqlConnection("Data Source=ProductHost;Initial Catalog=yourDB;Integrated Security=SSPI;"))
{
dbConnection.Open();
using (SqlBulkCopy s = new SqlBulkCopy(dbConnection))
{
s.DestinationTableName = "Your table name";
foreach (var column in csvFileData.Columns)
s.ColumnMappings.Add(column.ToString(), column.ToString());
s.WriteToServer(csvFileData);
}
}
【讨论】：
Microsoft.VisualBasic.FileIO does not exist，请点击this link。
我发现将大型 CSV 文件导入 SQL Server 的最佳方法是使用 SqlBulkCopy 和 IDataReader 实现。这样做的好处是您没有将整个文件读入内存（使用 DataTable 方法就是这种情况），您可以控制发送到 SQL Server 的批处理的大小。不好的是你必须实现IDataReader，这是我见过的最长的MS接口之一。
我编写了一个 nuget 包，可以为您解决问题。它使用了很棒的CsvHelper 包，因此需要的配置很少。最简单的场景如下所示：
//Instantiate the reader, providing the list of columns which matches 1 to 1 the data table structure.
var dataReader = new CsvDataReader(filePath,
new List<TypeCode>(5)
{
TypeCode.String,
TypeCode.Decimal,
TypeCode.String,
TypeCode.Boolean,
TypeCode.DateTime
});
bulkCopyUtility.BulkCopy("TableName", dataReader);
还有针对更复杂场景的其他配置选项（灵活的列映射、csv 文件中不存在的其他静态列值、值转换）。 如果您有兴趣，该项目位于Github，并以nuget package 的形式提供。
作为参考，下面是如何使用SqlBulkCopy 和IDataReader：
public void BulkCopy(string tableName, IDataReader dataReader, Action<SqlBulkCopy> configureSqlBulkCopy)
{
using (SqlConnection dbConnection = new SqlConnection(connectionString))
{
dbConnection.Open();
using (SqlBulkCopy bulkCopy = new SqlBulkCopy(dbConnection))
{
bulkCopy.BatchSize = 3000; //Data will be sent to SQL Server in batches of this size
bulkCopy.EnableStreaming = true;
bulkCopy.DestinationTableName = tableName;
//This will ensure mapping based on names rather than column position
foreach (DataColumn column in dataReader.GetSchemaTable().Columns)
{
bulkCopy.ColumnMappings.Add(column.ColumnName, column.ColumnName);
}
//If additional, custom configuration is required, invoke the action
configureSqlBulkCopy?.Invoke(bulkCopy);
try
{
// Write from the source to the destination.
bulkCopy.WriteToServer(dataReader);
}
finally
{
dataReader.Close();
}
}
}
}
【讨论】：
这是另一种通过编程将 CSV 上传到数据库的方法。 Cinchoo ETL，一个开源库可以使用 DataReader 完成将 CSV 文件上传到数据库的工作，这是一种几乎没有内存开销的流式方法。
这里是如何做的示例
string connectionstring = @"#YOUR DB ConnectionString#";
using (SqlBulkCopy bcp = new SqlBulkCopy(connectionstring))
{
using (var p = new ChoCSVReader("#YOUR CSV FILE#").WithFirstLineHeader())
{
bcp.DestinationTableName = "#TABLENAME#";
bcp.EnableStreaming = true;
bcp.BatchSize = 10000;
bcp.BulkCopyTimeout = 0;
bcp.NotifyAfter = 100;
bcp.SqlRowsCopied += delegate (object sender, SqlRowsCopiedEventArgs e)
{
Console.WriteLine(e.RowsCopied.ToString("#,##0") + " rows copied.");
};
bcp.WriteToServer(p.AsDataReader());
}
}
【讨论】：
load CSV with new headers dynamically to DB, create a new table with all the headers from the CSV 上面的代码有助于解决这个问题，但在同一个调用中我也想从表中取回一个 IList，所以在第 2 步中我怎样才能得到 @ 987654325@ 如果我们也能加载它会很棒
使用System.Data.SqlClient.SqlBulkCopy 类将数据插入到Sql 表中。 要使用该类，您还需要将 CVS 数据转换为 DataTable，请参阅here 方法之一。
【讨论】：
对于 SSIS 来说，这听起来像是一份完美的工作。它是 SQL Server 的免费部分，可以循环遍历文件夹中的所有 csv 文件，速度非常快，并且具有出色的错误处理和日志记录。
【讨论】：
此技术使用 SQLBulkCopy() 工具，但不会将整个文件读入内存。
诀窍在于它实现了一个 IDataReader 类来读取 .csv 文件。
https://www.codeproject.com/Tips/1029831/Fast-and-Simple-IDataReader-Implementation-to-Read
【讨论】：
你也可以使用Bulk Insert
Public Shared Function bulkQuery()
Dim query As StringBuilder = New StringBuilder
query.Append("USE Import_DB BULK INSERT dbo.[Insert_Table] FROM")
query.Append(" 'C:\Insert_Table.csv' ")
query.Append("With (FIELDTERMINATOR = ',', ROWTERMINATOR = '\n')")
Return query.ToString
End Function
这里要小心，因为表名和 csv 名称必须相同，而且 csv 中的列数必须与预定义表中的相同。
【讨论】：
private void GetDataTabletFromCSVFile(string fileName)
{
DataTable dt = new DataTable();
//dt.TableName = fileName;
try
{
using (TextFieldParser csvReader = new TextFieldParser(fileName))
{
csvReader.SetDelimiters(new string[] { "," });
csvReader.HasFieldsEnclosedInQuotes = true;
string[] colFields = csvReader.ReadFields();
//foreach (string column in colFields)
//{
// DataColumn datecolumn = new DataColumn(column);
// datecolumn.AllowDBNull = true;
// dt.Columns.Add(datecolumn);
//}
dt.Columns.AddRange(new DataColumn[8] {
new DataColumn("Symbol", typeof(string)),
new DataColumn("ISIN", typeof(string)),
new DataColumn("Company", typeof(string)),
new DataColumn("FirstListingDate", typeof(string)),
new DataColumn("FaceValue", typeof(string)),
new DataColumn("PaidUpValue", typeof(string)),
new DataColumn("MarketLot",typeof(string)),
new DataColumn("industry",typeof(string))
});
while (!csvReader.EndOfData)
{
string[] fieldData = csvReader.ReadFields();
//Making empty value as null
for (int i = 0; i < fieldData.Length; i++)
{
if (fieldData[i] == "")
{
fieldData[i] = null;
}
}
dt.Rows.Add(fieldData);
}
var builder = new ConfigurationBuilder()
.SetBasePath(Path.GetDirectoryName(System.Reflection.Assembly.GetExecutingAssembly().Location))
.AddJsonFile("appsettings.json");
var configuration = builder.Build();
string DBconnection = configuration.GetSection("ConnectionString").Value;
using (SqlConnection dbConnection = new SqlConnection(DBconnection))
{
dbConnection.Open();
using (SqlBulkCopy s = new SqlBulkCopy(dbConnection))
{
s.DestinationTableName = "Static.dbo.Securitiesinfo";
foreach (var column in dt.Columns)
s.ColumnMappings.Add(column.ToString(), column.ToString());
s.WriteToServer(dt);
}
}
}
}
catch (Exception ex)
{
var x = ex;
}
}
【讨论】：