Читайте из StreamReader в пакетном режиме

Question

Я столкнулся с исключениями OutOfMemory при попытке загрузить текстовый файл объемом 800 МБ в DataTable через StreamReader.Мне было интересно, есть ли способ загрузить DataTable из потока памяти в пакетном режиме, то есть прочитать первые 10000 строк текстового файла из StreamReader, создать DataTable, сделать что-то с DataTable, затем загрузить следующие 10000 строк в StreamReader искоро.

Мои Google не очень помогли, но, похоже, для этого должен быть простой способ.В конечном итоге я буду записывать DataTables в базу данных MS SQL с использованием SqlBulkCopy, поэтому, если есть более простой подход, чем тот, который я описал, я был бы благодарен за быстрый указатель в правильном направлении.

Правка - Воткод, который я запускаю:

public static DataTable PopulateDataTableFromText(DataTable dt, string txtSource)
{

    StreamReader sr = new StreamReader(txtSource);
    DataRow dr;
    int dtCount = dt.Columns.Count;
    string input;
    int i = 0;

    while ((input = sr.ReadLine()) != null)
    {

        try
        {
            string[] stringRows = input.Split(new char[] { '\t' });
            dr = dt.NewRow();
            for (int a = 0; a < dtCount; a++)
            {
                string dataType = dt.Columns[a].DataType.ToString();
                if (stringRows[a] == "" && (dataType == "System.Int32" || dataType == "System.Int64"))
                {
                    stringRows[a] = "0";
                }
                dr[a] = Convert.ChangeType(stringRows[a], dt.Columns[a].DataType);

            }
            dt.Rows.Add(dr);
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex.ToString());
        }
        i++;
    }
    return dt;
}

И вот возвращаемая ошибка:

«System.OutOfMemoryException: исключение типа« System.OutOfMemoryException »было сгенерировано.
в System.String.Split (разделитель Char [], число Int32, параметры StringSplitOptions)
в System.String.Split (разделитель Char []}
в Harvester.Config.PopulateDataTableFromText (DataTable dt, StringtxtSource) в C: .... "

Что касается предложения загружать данные непосредственно в SQL - я немного новичок, когда дело доходит до C #, но я подумал, что это в основном то, чтоЯ делаю? SqlBulkCopy.WriteToServer берет DataTable, который я создаю, из текстового файла и импортирует его в sql. Есть ли еще более простой способ сделать это, что мне не хватает?

Edit: О, я забыл упомянуть - этот код не будет работать на том же сервере, что и SQL Server.Текстовый файл данных находится на сервере B и должен быть записан в таблицу на сервере A. Это исключает использование bcp?

Answer 1

Рассматривали ли вы загрузку данных непосредственно в SQL Server и затем манипулирование ими в базе данных? Механизм базы данных уже разработан для эффективного управления большими объемами данных.Это может привести к лучшим результатам в целом и позволяет использовать возможности базы данных и языка SQL для выполнения тяжелой работы.Это старый принцип «работай умнее, а не усерднее» .

Существует несколько различных способов загрузки данных в SQL Server , поэтому вы можете изучить этичтобы увидеть, если они подходят.Если вы используете SQLServer 2005 или более поздней версии, и вам действительно необходимо выполнить некоторые манипуляции с данными в C #, вы всегда можете использовать управляемую хранимую процедуру .

Здесь нужно понять, чтоOutOfMemoryException немного вводит в заблуждение. Память - это больше, чем просто объем физической оперативной памяти, который у вас есть .Вероятно, вам не хватает адресуемой памяти .Это совсем другая вещь.

Когда вы загружаете большой файл в память и преобразуете его в DataTable, для представления одинаковых данных, вероятно, потребуется намного больше, чем просто 800 МБ. Поскольку 32-битные процессы .NET ограниченыдо чуть менее 2 ГБ адресуемой памяти, вы, вероятно, никогда не сможете обработать это количество данных в одном пакете.

Что вам, вероятно, потребуется, это обработать данные в потоковом режиме. Другими словами, не пытайтесь загрузить все это в DataTable и затем выполнить массовую вставку в SQLServer.Вместо этого обработайте файл кусками, очистив предыдущий набор строк, как только вы закончите с ними.

Теперь, если у вас есть доступ к 64-разрядной машине с большим объемом памяти (чтобы избежать перегрузки виртуальной машины)и копию 64-битной среды выполнения .NET, которую вы, вероятно, могли бы избежать при запуске кода без изменений.Но я бы предложил внести необходимые изменения в любом случае, поскольку это, вероятно, улучшит производительность даже в этой среде.

Answer 2

Вам действительно нужно обрабатывать данные пакетами строк? Или вы могли бы обрабатывать это построчно? В последнем случае я думаю, что Linq может быть очень полезен здесь, потому что он позволяет легко передавать данные через «конвейер» методов. Таким образом, вам не нужно загружать много данных одновременно, только одну строку за раз

Во-первых, вы должны сделать ваш StreamReader перечислимым. Это легко сделать с помощью метода расширения:

public static class TextReaderExtensions
{
    public static IEnumerable<string> Lines(this TextReader reader)
    {
        string line;
        while((line = reader.ReadLine()) != null)
        {
            yield return line;
        }
    }
}

Таким образом, вы можете использовать StreamReader в качестве источника для запроса Linq.

Тогда вам нужен метод, который берет строку и преобразует ее в DataRow:

DataRow ParseDataRow(string input)
{
    // Your parsing logic here
    ...
}

С этими элементами вы можете легко проецировать каждую строку из файла в DataRow и делать с ней все, что вам нужно:

using (var reader = new StreamReader(fileName))
{
    var rows = reader.Lines().Select(ParseDataRow);
    foreach(DataRow row in rows)
    {
        // Do something with the DataRow
    }
}

(обратите внимание, что вы можете сделать что-то похожее с простым циклом, без использования Linq, но я думаю, что Linq делает код более читабельным ...)

Answer 3

SqlBulkCopy.WriteToServer имеет перегрузку, которая принимает IDataReader.Вы можете реализовать свой собственный IDataReader в качестве оболочки для StreamReader, где метод Read () будет использовать одну строку из StreamReader.Таким образом, данные будут «перетекать» в базу данных вместо того, чтобы пытаться создать их в памяти как DataTable.Надеюсь, это поможет.

Answer 4

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

Суть кода в этом цикле:

//Of note: it's faster to read all the lines we are going to act on and 
            //then process them in parallel instead of reading and processing line by line.
            //Code source: http://cc.davelozinski.com/code/c-sharp-code/read-lines-in-batches-process-in-parallel
            while (blnFileHasMoreLines)
            {
                batchStartTime = DateTime.Now;  //Reset the timer

                //Read in all the lines up to the BatchCopy size or
                //until there's no more lines in the file
                while (intLineReadCounter < BatchSize && !tfp.EndOfData)
                {
                    CurrentLines[intLineReadCounter] = tfp.ReadFields();
                    intLineReadCounter += 1;
                    BatchCount += 1;
                    RecordCount += 1;
                }

                batchEndTime = DateTime.Now;    //record the end time of the current batch
                batchTimeSpan = batchEndTime - batchStartTime;  //get the timespan for stats

                //Now process each line in parallel.
                Parallel.For(0, intLineReadCounter, x =>
                //for (int x=0; x < intLineReadCounter; x++)    //Or the slower single threaded version for debugging
                {
                    List<object> values = null; //so each thread gets its own copy. 

                    if (tfp.TextFieldType == FieldType.Delimited)
                    {
                        if (CurrentLines[x].Length != CurrentRecords.Columns.Count)
                        {
                            //Do what you need to if the number of columns in the current line
                            //don't match the number of expected columns
                            return; //stop now and don't add this record to the current collection of valid records.
                        }

                        //Number of columns match so copy over the values into the datatable
                        //for later upload into a database
                        values = new List<object>(CurrentRecords.Columns.Count);
                        for (int i = 0; i < CurrentLines[x].Length; i++)
                            values.Add(CurrentLines[x][i].ToString());

                        //OR do your own custom processing here if not using a database.
                    }
                    else if (tfp.TextFieldType == FieldType.FixedWidth)
                    {
                        //Implement your own processing if the file columns are fixed width.
                    }

                    //Now lock the data table before saving the results so there's no thread bashing on the datatable
                    lock (oSyncLock)
                    {
                        CurrentRecords.LoadDataRow(values.ToArray(), true);
                    }

                    values.Clear();

                }
                ); //Parallel.For   

                //If you're not using a database, you obviously won't need this next piece of code.
                if (BatchCount >= BatchSize)
                {   //Do the SQL bulk copy and save the info into the database
                    sbc.BatchSize = CurrentRecords.Rows.Count;
                    sbc.WriteToServer(CurrentRecords);

                    BatchCount = 0;         //Reset these values
                    CurrentRecords.Clear(); //  "
                }

                if (CurrentLines[intLineReadCounter] == null)
                    blnFileHasMoreLines = false;    //we're all done, so signal while loop to stop

                intLineReadCounter = 0; //reset for next pass
                Array.Clear(CurrentLines, 0, CurrentLines.Length);

            } //while blnhasmorelines