Обработка больших данных на C#: от LINQ до параллельных потоков

Онлайн-тренажер Python для начинающих

Изучайте Python легко и без перегрузки теорией. Решайте практические задачи с автоматической проверкой, получайте подсказки на русском языке и пишите код прямо в браузере — без необходимости что-либо устанавливать.

Начать курс

Почему C# — ваш секретный инструмент для Big Data?

Когда речь заходит об обработке больших данных, в голову сразу приходят Python с его Pandas, Scala с Spark или Java с Hadoop. Но давайте честно: C# часто незаслуженно остаётся в тени. А зря. Платформа .NET даёт нам такие мощные инструменты, как PLINQ, MemoryMappedFile, Parallel.For и асинхронные потоки, которые позволяют обрабатывать гигабайты данных на обычном железе без танцев с бубном.

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

Главная проблема: как не убить память?

Представьте: у вас есть CSV-файл на 50 ГБ с логами сервера. Вы открываете его в обычном StreamReader и читаете построчно. Всё хорошо? Нет. Если вы попытаетесь загрузить его целиком в List<string> — привет, OutOfMemoryException. Даже если у вас 64 ГБ RAM, .NET может фрагментировать память так, что большой объект просто не поместится в кучу.

Решение номер один: стриминг и чанкование. Не загружайте всё сразу — жуйте по кусочкам.

public async IAsyncEnumerable<string[]> ReadCsvChunksAsync(string filePath, int chunkSize = 10000)
{
    using var reader = new StreamReader(filePath);
    string? line;
    var buffer = new List<string>(chunkSize);

    while ((line = await reader.ReadLineAsync()) != null)
    {
        buffer.Add(line);
        if (buffer.Count >= chunkSize)
        {
            yield return buffer.ToArray();
            buffer.Clear();
        }
    }

    if (buffer.Count > 0)
        yield return buffer.ToArray();
}

Теперь вы можете обрабатывать гигабайты, не думая о памяти:

await foreach (var chunk in ReadCsvChunksAsync("logs.csv"))
{
    ProcessChunk(chunk); // ваша логика
}

Параллельная обработка: используем все ядра процессора

Хорошо, мы читаем данные порциями. Но если у вас 16-ядерный процессор, а данные обрабатываются последовательно — вы используете только 6% его мощности. Пора включать параллельные вычисления на C#.

PLINQ — Parallel LINQ

Самая простая магия — добавить .AsParallel() к вашему LINQ-запросу. Но с большими данными нужно быть аккуратнее: PLINQ сам решает, когда распараллеливать, а это может привести к излишней фрагментации.

var numbers = Enumerable.Range(1, 10_000_000);

// Обычный LINQ (медленно)
var sum = numbers.Where(n => n % 2 == 0).Sum();

// PLINQ (быстро, но осторожно)
var parallelSum = numbers
    .AsParallel()
    .WithDegreeOfParallelism(Environment.ProcessorCount)
    .Where(n => n % 2 == 0)
    .Sum();

Console.WriteLine($"Результат: {parallelSum}");

Но PLINQ не панацея. Для сложной обработки больших данных на C# лучше использовать Parallel.ForEach с ручным управлением разбивкой.

Parallel.ForEach с разбивкой на диапазоны

var allLines = File.ReadLines("huge_dataset.csv");
var processedCount = 0;
var lockObject = new object();

Parallel.ForEach(
    allLines,
    new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount },
    line =>
    {
        var result = ProcessLine(line); // ваша тяжёлая функция
        lock (lockObject)
        {
            processedCount++;
            if (processedCount % 100_000 == 0)
                Console.WriteLine($"Обработано: {processedCount}");
        }
    });

MemoryMappedFile: работаем с файлами как с оперативной памятью

Когда файл слишком большой для загрузки в RAM, но вам нужен произвольный доступ к его частям, на помощь приходит MemoryMappedFile. Это технология, которая позволяет отобразить часть файла в виртуальное адресное пространство процесса. Операционная система сама решает, какие страницы держать в физической памяти.

Пример: у вас есть бинарный файл с записями фиксированной длины (например, 256 байт каждая). Вам нужно прочитать запись №1 000 000.

using System.IO.MemoryMappedFiles;

public class BinaryRecordReader : IDisposable
{
    private readonly MemoryMappedFile _mmf;
    private readonly long _recordSize;

    public BinaryRecordReader(string filePath, long recordSize)
    {
        _recordSize = recordSize;
        _mmf = MemoryMappedFile.CreateFromFile(filePath, FileMode.Open);
    }

    public byte[] ReadRecord(long index)
    {
        var offset = index * _recordSize;
        using var accessor = _mmf.CreateViewAccessor(offset, _recordSize);
        var buffer = new byte[_recordSize];
        accessor.ReadArray(0, buffer, 0, buffer.Length);
        return buffer;
    }

    public void Dispose()
    {
        _mmf?.Dispose();
    }
}

// Использование
using var reader = new BinaryRecordReader("bigdata.bin", 256);
var record = reader.ReadRecord(1_000_000);
Console.WriteLine($"Прочитана запись длиной {record.Length} байт");

Это невероятно быстро — никакого копирования в буфер, никаких дисковых операций на уровне C#. Всё делает ОС.

Асинхронные потоки: IAsyncEnumerable для Big Data

C# 8.0 принёс нам асинхронные потоки (IAsyncEnumerable<T>). Это идеальный паттерн для обработки больших данных: вы читаете и обрабатываете данные по мере их поступления, не блокируя поток UI или веб-запроса.

Реальный пример: обработка JSON-логов с API. Допустим, у вас есть файл, где каждая строка — это JSON-объект.

using System.Text.Json;

public record LogEntry(DateTime Timestamp, string Level, string Message);

public async IAsyncEnumerable<LogEntry> ReadLogsAsync(string filePath)
{
    using var reader = new StreamReader(filePath);
    while (await reader.ReadLineAsync() is { } line)
    {
        if (string.IsNullOrWhiteSpace(line)) continue;

        LogEntry? entry = null;
        try
        {
            entry = JsonSerializer.Deserialize<LogEntry>(line);
        }
        catch (JsonException)
        {
            // логируем ошибку, но не падаем
        }

        if (entry != null)
            yield return entry;
    }
}

// Использование с фильтрацией на лету
await foreach (var log in ReadLogsAsync("app.log.json"))
{
    if (log.Level == "Error")
    {
        Console.WriteLine($"[{log.Timestamp}] {log.Message}");
    }
}