C#C
C#•3y ago
engineertdog

BulkInsert to SQL, parallel processing

I'm trying to improve the processing speed of a 3rd party integration (via SDK), wherein the processed data is written to SQL.

I was able to reduce the total time of processing the 3rd party system from 40 hours to 3 by improving the methods used to process the data, and to turn the sequencing processing into parallel processing. The challenge I face now is with writing the data to SQL.

What I currently have is the following:
c#
private ConcurrentBag<DataRow> _dataTableRows;
private DataTable _dataTable;

IList<PIPoint> points = PIPoint.FindPIPoints(_piServer, pointQuery).ToList();
PIPointList pointValues = new PIPointList(points);
AFListResults<PIPoint, AFValue> values = pointValues.CurrentValue();

Parallel.ForEach(values, value => {
  IDcitionary<string, object> pointAttributes = value.PIPoint.GetAttributes();
  DataRow row = ConfigureRow(pointAttributes, value); // this simply converts the 55 attributes & value (with timestamp) into the string columns for the data row

  _dataTableRows.Add(row);
  this.AddDataTableRecords(100000);
});

public void AddDataTableRecords(int threshold) {
  if (_dataTableRows.Count >= threshold) {
    lock (_dataTable.Rows.SyncRoot) {
      foreach (DataRow row in _dataTableRows.Take(_dataTableRows.Count) {
        _dataTable.Rows.Add(row);
      }

      bool success = _dataController.UpsertDataTableRecords(_dataTable);
    
      if (success) {
        _dataTable.Rows.Clear();
      }
    }
  }
}


With this method, I'll use a temp table with a merge / insert strategy to add or update the records. The problem is that the original Parallel.ForEach is sometimes running twice for the same record, when there truly is only 1 record available. There's 8 million records I need to process in this manner, and without using the Parallel method on the input records, it takes far too long to process all the data. Any thoughts / help?
Was this page helpful?