Как улучшить производительность загрузки CSV через datatable

Question

У меня есть рабочее решение для загрузки файла CSV. В настоящее время я использую IFormCollection для пользователя, чтобы загрузить несколько файлов CSV из представления.

Файлы CSV сохраняются в виде временного файла следующим образом:

List<string> fileLocations = new List<string>();
foreach (var formFile in files)
{
   filePath = Path.GetTempFileName();    
   if (formFile.Length > 0)
   {
       using (var stream = new FileStream(filePath, FileMode.Create))
       {
           await formFile.CopyToAsync(stream);
       }
   }

   fileLocations.Add(filePath);
}

Я отправляю список расположений файлов другому методу (чуть ниже). Я перебираю расположение файлов и передаю данные из временных файлов, затем использую таблицу данных и SqlBulkCopy для вставки данных. В настоящее время я загружаю от 50 до 200 файлов одновременно, и каждый файл составляет около 330 КБ. Чтобы вставить сотню, требуется около 6 минут, что составляет около 30-35 МБ.

public void SplitCsvData(string fileLocation, Guid uid)
        {
            MetaDataModel MetaDatas;
            List<RawDataModel> RawDatas;

            var reader = new StreamReader(File.OpenRead(fileLocation));
            List<string> listRows = new List<string>();
            while (!reader.EndOfStream)
            {
                listRows.Add(reader.ReadLine());
            }

            var metaData = new List<string>();
            var rawData = new List<string>();

            foreach (var row in listRows)
            {
                var rowName = row.Split(',')[0];
                bool parsed = int.TryParse(rowName, out int result);

                if (parsed == false)
                {
                    metaData.Add(row);
                }
                else
                {
                    rawData.Add(row);
                }
            }

         //Assigns the vertical header name and value to the object by splitting string 
         RawDatas = GetRawData.SplitRawData(rawData);
         SaveRawData(RawDatas);

         MetaDatas = GetMetaData.SplitRawData(rawData);
         SaveRawData(RawDatas);

        }

Затем этот код передает объект в объект для создания таблицы данных и вставки данных.

private DataTable CreateRawDataTable
{
   get
   {
       var dt = new DataTable();
       dt.Columns.Add("Id", typeof(int));
       dt.Columns.Add("SerialNumber", typeof(string));
       dt.Columns.Add("ReadingNumber", typeof(int));
       dt.Columns.Add("ReadingDate", typeof(string));
       dt.Columns.Add("ReadingTime", typeof(string));
       dt.Columns.Add("RunTime", typeof(string));
       dt.Columns.Add("Temperature", typeof(double));
       dt.Columns.Add("ProjectGuid", typeof(Guid));
       dt.Columns.Add("CombineDateTime", typeof(string));

        return dt;
  }
}

public void SaveRawData(List<RawDataModel> data)
{
       DataTable dt = CreateRawDataTable;
       var count = data.Count;          

       for (var i = 1; i < count; i++)
       {
           DataRow row = dt.NewRow();
           row["Id"] = data[i].Id;
           row["ProjectGuid"] = data[i].ProjectGuid;
           row["SerialNumber"] = data[i].SerialNumber;
           row["ReadingNumber"] = data[i].ReadingNumber;
           row["ReadingDate"] = data[i].ReadingDate;
           row["ReadingTime"] = data[i].ReadingTime;
           row["CombineDateTime"] = data[i].CombineDateTime;
           row["RunTime"] = data[i].RunTime;
           row["Temperature"] = data[i].Temperature;
           dt.Rows.Add(row);
        }

        using (var conn = new SqlConnection(connectionString))
        {
           conn.Open();
           using (SqlTransaction tr = conn.BeginTransaction())
           {
               using (var sqlBulk = new SqlBulkCopy(conn, SqlBulkCopyOptions.Default, tr))
               {
                   sqlBulk.BatchSize = 1000;
                   sqlBulk.DestinationTableName = "RawData";
                   sqlBulk.WriteToServer(dt);
               }
               tr.Commit();
           }
       }
   }

Есть ли другой способ сделать это или более эффективный способ повысить производительность, чтобы сократить время загрузки, поскольку это может занять много времени, и я вижу, что использование памяти увеличивается до 500 МБ.

ТИА

Answer 1

В дополнение к ответу @ Panagiotis, почему бы вам не чередовать обработку файлов с загрузкой файлов? Завершите свою логику обработки файлов в асинхронном методе и измените цикл на Parallel.Foreach и обрабатывайте каждый файл по мере его поступления, вместо того чтобы ждать их всех?

private static readonly object listLock = new Object(); // only once at class level


    List<string> fileLocations = new List<string>();
    Parallel.ForEach(files, (formFile) => 
    {
       filePath = Path.GetTempFileName();    
       if (formFile.Length > 0)
       {
           using (var stream = new FileStream(filePath, FileMode.Create))
           {
               await formFile.CopyToAsync(stream);
           }

           await ProcessFileInToDbAsync(filePath); 
       }

       // Added lock for thread safety of the List 
       lock (listLock)
       {
           fileLocations.Add(filePath);
       }     
    });

Answer 2

Благодаря @Panagiotis Kanavos я смог понять, что делать. Во-первых, способ, которым я вызывал методы, оставлял их в памяти. У меня есть CSV-файл, состоящий из двух частей: вертикальные метаданные и обычная горизонтальная информация. Поэтому мне нужно было разделить их на две части. Сохранение их в виде файлов tmp также вызывало накладные расходы. Это заняло от 5-6 минут до минуты, что, на мой взгляд, для 100 файлов, содержащих 8500 строк, неплохо.

Вызов метода:

public async Task<IActionResult> UploadCsvFiles(ICollection<IFormFile> files, IFormCollection fc)
{
   foreach (var f in files)
   {
       var getData = new GetData(_configuration);
       await getData.SplitCsvData(f, uid);
   }

   return whatever;
}

Это метод, выполняющий разбиение:

public async Task SplitCsvData(IFormFile file, string uid)
    {
        var data = string.Empty;
        var m = new List<string>();
        var r = new List<string>();

        var records = new List<string>();
        using (var stream = file.OpenReadStream())
        using (var reader = new StreamReader(stream))
        {
            while (!reader.EndOfStream)
            {
                var line = reader.ReadLine();
                var header = line.Split(',')[0].ToString();
                bool parsed = int.TryParse(header, out int result);
                if (!parsed)
                {
                    m.Add(line);
                }
                else
                {
                    r.Add(line);
                }
            }
        }

        //TODO: Validation
        //This splits the list into the Meta data model. This is just a single object, with static fields.
        var metaData = SplitCsvMetaData.SplitMetaData(m, uid);
        DataTable dtm = CreateMetaData(metaData);
        var serialNumber = metaData.LoggerId;
        await SaveMetaData("MetaData", dtm);

        //
        var lrd = new List<RawDataModel>();
        foreach (string row in r)
        {
            lrd.Add(new RawDataModel
            {
                Id = 0,
                SerialNumber = serialNumber,
                ReadingNumber = Convert.ToInt32(row.Split(',')[0]),
                ReadingDate = Convert.ToDateTime(row.Split(',')[1]).ToString("yyyy-MM-dd"),
                ReadingTime = Convert.ToDateTime(row.Split(',')[2]).ToString("HH:mm:ss"),
                RunTime = row.Split(',')[3].ToString(),
                Temperature = Convert.ToDouble(row.Split(',')[4]),
                ProjectGuid = uid.ToString(),
                CombineDateTime = Convert.ToDateTime(row.Split(',')[1] + " " + row.Split(',')[2]).ToString("yyyy-MM-dd HH:mm:ss")
            });
        }

        await SaveRawData("RawData", lrd);
    }

Затем я использую таблицу данных для метаданных (которая занимает 100 секунд для 100 файлов), когда я сопоставляю имена полей со столбцами.

 public async Task SaveMetaData(string table, DataTable dt)
    {
        using (SqlBulkCopy sqlBulk = new SqlBulkCopy(_configuration.GetConnectionString("DefaultConnection"), SqlBulkCopyOptions.Default))
        { 
            sqlBulk.DestinationTableName = table;
            await sqlBulk.WriteToServerAsync(dt);
        }
    }

Затем я использую FastMember для больших частей данных для необработанных данных, что больше похоже на традиционный CSV.

 public async Task SaveRawData(string table, IEnumerable<LogTagRawDataModel> lrd)
    {
        using (SqlBulkCopy sqlBulk = new SqlBulkCopy(_configuration.GetConnectionString("DefaultConnection"), SqlBulkCopyOptions.Default))
        using (var reader = ObjectReader.Create(lrd, "Id","SerialNumber", "ReadingNumber", "ReadingDate", "ReadingTime", "RunTime", "Temperature", "ProjectGuid", "CombineDateTime"))
        {                
            sqlBulk.DestinationTableName = table;
            await sqlBulk.WriteToServerAsync(reader);
        }  
    }

Я уверен, что это можно улучшить, но сейчас это работает очень хорошо.

Answer 3

Вы можете улучшить производительность, удалив DataTable и считав непосредственно из потока ввода.

SqlBulkCopy имеет перегрузку WriteToServer , которая принимает IDataReader вместо всей таблицы данных.

CsvHelper может CSV-файлы, используя StreamReader в качестве входных данных. Он предоставляет CsvDataReader как IDataReader реализацию поверх данных CSV. Это позволяет читать непосредственно из входного потока и записывать в SqlBulkCopy.

Следующий метод будет читать из IFormFile, анализировать поток с помощью CsvHelper и использовать поля CSV для настройки экземпляра SqlBulkCopy:

public async Task ToTable(IFormFile file, string table)
{
    using (var stream = file.OpenReadStream())
    using (var tx = new StreamReader(stream))
    using (var reader = new CsvReader(tx))
    using (var rd = new CsvDataReader(reader))
    {
        var headers = reader.Context.HeaderRecord;

        var bcp = new SqlBulkCopy(_connection)
        {
            DestinationTableName = table
        };
        //Assume the file headers and table fields have the same names
        foreach(var header in headers)
        {
            bcp.ColumnMappings.Add(header, header);
        }

        await bcp.WriteToServerAsync(rd);                
    }
}

Таким образом, ничто никогда не записывается во временную таблицу или не кэшируется в памяти. Загруженные файлы анализируются и записываются в базу данных напрямую.