Xuất tệp CSV lớn song song với máy chủ SQL

Question

Tôi có tệp CSV lớn ... 10 cột, 100 triệu hàng, có kích thước khoảng 6 GB trên đĩa cứng của tôi. Tôi muốn đọc dòng tệp CSV này và sau đó tải dữ liệu vào cơ sở dữ liệu máy chủ Microsoft SQL bằng cách sử dụng bản sao số lượng lớn SQL. Tôi đã đọc vài chủ đề ở đây và trên internet. Hầu hết mọi người đề nghị rằng đọc một tập tin CSV song song không mua nhiều về hiệu quả như các nhiệm vụ/chủ đề tranh giành quyền truy cập đĩa.

Điều tôi đang cố gắng làm là đọc từng dòng một từ CSV và thêm nó vào để chặn thu thập kích thước 100 nghìn hàng. Và một khi bộ sưu tập này đầy đủ quay lên một nhiệm vụ/luồng mới để ghi dữ liệu vào máy chủ SQL bằng cách sử dụng API SQLBuckCopy.

Tôi đã viết đoạn mã này, nhưng nhấn một lỗi vào thời gian chạy có nội dung "Cố gắng gọi bản sao hàng loạt trên một đối tượng có hoạt động đang chờ xử lý". Kịch bản này trông giống như một cái gì đó mà có thể dễ dàng giải quyết bằng cách sử dụng NET 4.0 TPL nhưng tôi không thể làm cho nó hoạt động. Bất kỳ đề xuất về những gì tôi đang làm sai?

public static void LoadCsvDataInParalleToSqlServer(string fileName, string connectionString, string table, DataColumn[] columns, bool truncate) 
    { 
     const int inputCollectionBufferSize = 1000000; 
     const int bulkInsertBufferCapacity = 100000; 
     const int bulkInsertConcurrency = 8; 

     var sqlConnection = new SqlConnection(connectionString); 
     sqlConnection.Open(); 

     var sqlBulkCopy = new SqlBulkCopy(sqlConnection.ConnectionString, SqlBulkCopyOptions.TableLock) 
     { 
      EnableStreaming = true, 
      BatchSize = bulkInsertBufferCapacity, 
      DestinationTableName = table, 
      BulkCopyTimeout = (24 * 60 * 60), 
     }; 

     BlockingCollection<DataRow> rows = new BlockingCollection<DataRow>(inputCollectionBufferSize); 
     DataTable dataTable = new DataTable(table); 
     dataTable.Columns.AddRange(columns); 

     Task loadTask = Task.Factory.StartNew(() => 
      { 
       foreach (DataRow row in ReadRows(fileName, dataTable)) 
       { 
        rows.Add(row); 
       } 

       rows.CompleteAdding(); 
      }); 

     List<Task> insertTasks = new List<Task>(bulkInsertConcurrency); 

     for (int i = 0; i < bulkInsertConcurrency; i++) 
     { 
      insertTasks.Add(Task.Factory.StartNew((x) => 
       { 
        List<DataRow> bulkInsertBuffer = new List<DataRow>(bulkInsertBufferCapacity); 

        foreach (DataRow row in rows.GetConsumingEnumerable()) 
        { 
         if (bulkInsertBuffer.Count == bulkInsertBufferCapacity) 
         { 
          SqlBulkCopy bulkCopy = x as SqlBulkCopy; 
          var dataRows = bulkInsertBuffer.ToArray(); 
          bulkCopy.WriteToServer(dataRows); 
          Console.WriteLine("Inserted rows " + bulkInsertBuffer.Count); 
          bulkInsertBuffer.Clear(); 
         } 

         bulkInsertBuffer.Add(row); 
        } 

       }, 
       sqlBulkCopy)); 
     } 

     loadTask.Wait(); 
     Task.WaitAll(insertTasks.ToArray()); 
    } 

    private static IEnumerable<DataRow> ReadRows(string fileName, DataTable dataTable) 
    { 
     using (var textFieldParser = new TextFieldParser(fileName)) 
     { 
      textFieldParser.TextFieldType = FieldType.Delimited; 
      textFieldParser.Delimiters = new[] { "," }; 
      textFieldParser.HasFieldsEnclosedInQuotes = true; 

      while (!textFieldParser.EndOfData) 
      { 
       string[] cols = textFieldParser.ReadFields(); 

       DataRow row = dataTable.NewRow(); 

       for (int i = 0; i < cols.Length; i++) 
       { 
        if (string.IsNullOrEmpty(cols[i])) 
        { 
         row[i] = DBNull.Value; 
        } 
        else 
        { 
         row[i] = cols[i]; 
        } 
       } 

       yield return row; 
      } 
     } 
    } 

Answer 1

Không.

Truy cập song song có thể hoặc không cho phép bạn đọc nhanh hơn tệp (nó sẽ không, nhưng tôi sẽ không chiến đấu rằng trận chiến ...) nhưng đối với một số ghi song song, nó sẽ không cung cấp cho bạn chèn số lượng lớn nhanh hơn. Đó là do chèn số lượng lớn được ghi tối thiểu (ví dụ: thực sự nhanh chóng chèn số lượng lớn) yêu cầu khóa bàn. Xem Prerequisites for Minimal Logging in Bulk Import:

khai thác gỗ tối thiểu đòi hỏi rằng các bảng mục tiêu đáp ứng các điều kiện sau đây:

...
- Bảng khóa được quy định (sử dụng TABLOCK).
...

Chèn song song, theo định nghĩa, không thể có được ổ khóa đồng thời. QED. Bạn đang sủa cây sai.

Ngừng nhận nguồn của bạn từ tìm kiếm ngẫu nhiên trên internet. Đọc The Data Loading Performance Guide, là hướng dẫn để ... tải dữ liệu có hiệu suất.

Tôi khuyên bạn nên ngừng phát minh ra bánh xe. Sử dụng SSIS, đây là chính xác những gì được thiết kế để xử lý.

Answer 2

http://joshclose.github.io/CsvHelper/

https://efbulkinsert.codeplex.com/

Nếu có thể cho bạn, tôi đề nghị bạn đọc tập tin của bạn vào một danh sách < T> sử dụng csvhelper nói trên và ghi vào db của bạn sử dụng số lượng lớn chèn như bạn đang làm hoặc efbulkinsert mà tôi đã sử dụng và rất nhanh.

using CsvHelper; 

public static List<T> CSVImport<T,TClassMap>(string csvData, bool hasHeaderRow, char delimiter, out string errorMsg) where TClassMap : CsvHelper.Configuration.CsvClassMap 
    { 
     errorMsg = string.Empty; 
     var result = Enumerable.Empty<T>(); 

     MemoryStream memStream = new MemoryStream(Encoding.UTF8.GetBytes(csvData)); 
     StreamReader streamReader = new StreamReader(memStream); 
     var csvReader = new CsvReader(streamReader); 

     csvReader.Configuration.RegisterClassMap<TClassMap>(); 
     csvReader.Configuration.DetectColumnCountChanges = true; 
     csvReader.Configuration.IsHeaderCaseSensitive = false; 
     csvReader.Configuration.TrimHeaders = true; 
     csvReader.Configuration.Delimiter = delimiter.ToString(); 
     csvReader.Configuration.SkipEmptyRecords = true; 
     List<T> items = new List<T>(); 

     try 
     { 
      items = csvReader.GetRecords<T>().ToList(); 
     } 
     catch (Exception ex) 
     { 
      while (ex != null) 
      { 
       errorMsg += ex.Message + Environment.NewLine; 

       foreach (var val in ex.Data.Values) 
        errorMsg += val.ToString() + Environment.NewLine; 

       ex = ex.InnerException; 
      } 
     } 
     return items; 
    } 
} 

Chỉnh sửa - Tôi không hiểu bạn đang làm gì với chèn hàng loạt. Bạn muốn chèn số lượng lớn toàn bộ danh sách hoặc bảng dữ liệu dữ liệu, chứ không phải từng hàng.

Answer 3

Bạn có thể tạo thủ tục lưu trữ và vượt qua các vị trí tập tin như dưới đây

CREATE PROCEDURE [dbo].[CSVReaderTransaction] 
    @Filepath varchar(100)='' 
AS 
-- STEP 1: Start the transaction 
BEGIN TRANSACTION 

-- STEP 2 & 3: checking @@ERROR after each statement 
EXEC ('BULK INSERT Employee FROM ''' +@Filepath 
     +''' WITH (FIELDTERMINATOR = '','', ROWTERMINATOR = ''\n'')') 

-- Rollback the transaction if there were any errors 
IF @@ERROR <> 0 
BEGIN 
    -- Rollback the transaction 
    ROLLBACK 

    -- Raise an error and return 
    RAISERROR ('Error in inserting data into employee Table.', 16, 1) 
    RETURN 
END 

COMMIT TRANSACTION 

Bạn cũng có thể thêm tùy chọn BATCHSIZE như FIELDTERMINATOR và ROWTERMINATOR.