Высокоскоростные конвейеры ETL на C# с параллельной обработкой

// Параллельная обработка коллекции данных
Parallel.ForEach(dataItems, item => 
{
    // Трансформация элемента данных
    var transformedItem = TransformData(item);
    
    // Здесь может быть дополнительная обработка
});

async Task<ProcessedData> ProcessETLPipeline(RawData data)
{
    var extracted = await ExtractAsync(data);
    var transformed = await TransformAsync(extracted);
    return await LoadAsync(transformed);
}

BlockingCollection<DataItem> buffer = new BlockingCollection<DataItem>(boundedCapacity: 1000);
 
// Поток-производитель
Task.Run(() => {
    foreach (var item in sourceData)
        buffer.Add(item);
    buffer.CompleteAdding();
});
 
// Потоки-потребители
Parallel.For(0, Environment.ProcessorCount, i => {
    foreach (var item in buffer.GetConsumingEnumerable())
        ProcessItem(item);
});

// Фаза Map - разбиение и параллельная обработка
var mappedResults = dataChunks.AsParallel()
                              .SelectMany(chunk => Map(chunk))
                              .ToList();
 
// Фаза Reduce - объединение результатов
var finalResult = Reduce(mappedResults);

// Создание канала с ограниченной емкостью
var channel = Channel.CreateBounded<DataItem>(new BoundedChannelOptions(1000)
{
    FullMode = BoundedChannelFullMode.Wait // Блокировать при заполнении
});
 
// Производитель
async Task ProduceAsync()
{
    foreach (var item in dataSource)
    {
        await channel.Writer.WriteAsync(item);
    }
    channel.Writer.Complete();
}
 
// Потребитель
async Task ConsumeAsync()
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        await ProcessItemAsync(item);
    }
}
 
// Запуск обоих процессов параллельно
await Task.WhenAll(ProduceAsync(), ConsumeAsync());

// Источник данных
var extractBlock = new TransformBlock<string, RawData>(
    sourceUrl => ExtractDataFromSource(sourceUrl),
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 }
);
 
// Блок трансформации
var transformBlock = new TransformBlock<RawData, TransformedData>(
    rawData => TransformData(rawData),
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 8 }
);
 
// Блок загрузки
var loadBlock = new ActionBlock<TransformedData>(
    transformedData => LoadDataToDestination(transformedData),
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 2 }
);
 
// Соединение блоков в конвейер
extractBlock.LinkTo(transformBlock, new DataflowLinkOptions { PropagateCompletion = true });
transformBlock.LinkTo(loadBlock, new DataflowLinkOptions { PropagateCompletion = true });

var options = new ExecutionDataflowBlockOptions 
{
    BoundedCapacity = 100, // Ограничение буфера
    MaxDegreeOfParallelism = 4
};

// Определение секций для параллельного извлечения
var partitions = new[]
{
    "WHERE id BETWEEN 1 AND 1000000",
    "WHERE id BETWEEN 1000001 AND 2000000",
    "WHERE id BETWEEN 2000001 AND 3000000"
};
 
// Параллельное извлечение из разных секций
var extractTasks = partitions.Select(partition => 
    ExtractDataAsync($"SELECT * FROM large_table {partition}")
).ToArray();
 
// Ожидание завершения всех задач извлечения
await Task.WhenAll(extractTasks);

public class ParallelEtlPipeline
{
    private readonly Extractor _extractor;
    private readonly Transformer _transformer;
    private readonly Loader _loader;
    
    public ParallelEtlPipeline(string connectionString)
    {
        _extractor = new Extractor(connectionString);
        _transformer = new Transformer();
        _loader = new Loader(connectionString);
    }
    
    public async Task ExecuteAsync()
    {
        // Здесь будет основная логика конвейера
    }
}

public async Task ExecuteAsync(string extractQuery, string targetTable, int batchSize = 10000)
{
    // Извлечение данных
    var rawData = await _extractor.ExtractDataAsync(extractQuery);
    
    // Разбиение на пакеты для параллельной обработки
    var batches = CreateBatches(rawData, batchSize);
    
    // Параллельная трансформация
    var transformTasks = batches.Select(batch => 
        Task.Run(() => _transformer.TransformDataBatch(batch))
    ).ToArray();
    
    var transformedBatches = await Task.WhenAll(transformTasks);
    
    // Загрузка результатов (возможно тоже параллельная)
    foreach (var batch in transformedBatches)
    {
        await _loader.LoadDataAsync(batch, targetTable);
    }
}
 
private List<List<T>> CreateBatches<T>(List<T> items, int batchSize)
{
    return Enumerable.Range(0, (items.Count + batchSize - 1) / batchSize)
        .Select(i => items.Skip(i * batchSize).Take(batchSize).ToList())
        .ToList();
}

public async Task StreamingExecuteAsync(string extractQuery, string targetTable, int bufferSize = 1000)
{
    using var extractedItems = new BlockingCollection<RawData>(bufferSize);
    using var transformedItems = new BlockingCollection<TransformedData>(bufferSize);
    
    // Задача извлечения данных
    var extractTask = Task.Run(async () =>
    {
        await foreach (var item in _extractor.ExtractStreamAsync(extractQuery))
        {
            extractedItems.Add(item);
        }
        extractedItems.CompleteAdding();
    });
    
    // Задача трансформации данных
    var transformTask = Task.Run(() =>
    {
        Parallel.ForEach(extractedItems.GetConsumingEnumerable(), 
            new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount },
            rawItem =>
            {
                var transformedItem = _transformer.TransformItem(rawItem);
                transformedItems.Add(transformedItem);
            });
        
        transformedItems.CompleteAdding();
    });
    
    // Задача загрузки данных
    var loadTask = Task.Run(async () =>
    {
        var batchBuffer = new List<TransformedData>();
        foreach (var item in transformedItems.GetConsumingEnumerable())
        {
            batchBuffer.Add(item);
            
            // Загружаем данные пакетами для эффективности
            if (batchBuffer.Count >= 100)
            {
                await _loader.LoadDataBatchAsync(batchBuffer, targetTable);
                batchBuffer.Clear();
            }
        }
        
        // Загружаем остаток
        if (batchBuffer.Any())
            await _loader.LoadDataBatchAsync(batchBuffer, targetTable);
    });
    
    // Ждем завершения всех задач
    await Task.WhenAll(extractTask, transformTask, loadTask);
}

// Асинхронное извлечение данных потоком
public async IAsyncEnumerable<DataItem> ExtractDataStreamAsync(string source)
{
    using var connection = new SqlConnection(_connectionString);
    await connection.OpenAsync();
    
    using var command = new SqlCommand(source, connection);
    using var reader = await command.ExecuteReaderAsync();
    
    while (await reader.ReadAsync())
    {
        var item = new DataItem
        {
            Id = reader.GetInt32(0),
            Name = reader.GetString(1),
            // Другие поля...
        };
        
        yield return item;
    }
}

// Трансформация потока данных
public async IAsyncEnumerable<TransformedItem> TransformStreamAsync(IAsyncEnumerable<DataItem> items)
{
    await foreach (var item in items)
    {
        // Асинхронная трансформация - например, обогащение данными из внешнего сервиса
        var enrichedData = await _enrichmentService.EnrichAsync(item);
        
        yield return new TransformedItem
        {
            Id = item.Id,
            ProcessedName = item.Name.ToUpperInvariant(),
            EnrichmentData = enrichedData
            // Другие трансформированные поля...
        };
    }
}

public async Task ProcessStreamAsync(string source, string destination)
{
    var extractedItems = _extractor.ExtractDataStreamAsync(source);
    var transformedItems = _transformer.TransformStreamAsync(extractedItems);
    
    await _loader.LoadStreamAsync(transformedItems, destination);
}

// Параллельная обработка по диапазонам ID
async Task ProcessByIdRangesAsync(int totalRecords, int rangeSize)
{
    var ranges = Enumerable.Range(0, (totalRecords + rangeSize - 1) / rangeSize)
        .Select(i => new { Start = i * rangeSize + 1, End = Math.Min((i + 1) * rangeSize, totalRecords) })
        .ToList();
    
    var tasks = ranges.Select(async range => 
    {
        string query = $"SELECT * FROM source_table WHERE id BETWEEN {range.Start} AND {range.End}";
        var extractedItems = _extractor.ExtractDataAsync(query);
        // Дальнейшая обработка...
    });
    
    await Task.WhenAll(tasks);
}

async Task ProcessVerticalPartitioning(List<DataItem> items)
{
    // Параллельная обработка разных аспектов данных
    var nameProcessingTask = Task.Run(() => 
        items.AsParallel().Select(i => ProcessName(i.Name)).ToList());
    
    var amountProcessingTask = Task.Run(() => 
        items.AsParallel().Select(i => CalculateAmount(i.Amount)).ToList());
    
    // Ожидание завершения всех задач
    await Task.WhenAll(nameProcessingTask, amountProcessingTask);
    
    // Объединение результатов
    var processedNames = nameProcessingTask.Result;
    var calculatedAmounts = amountProcessingTask.Result;
    
    // Создание итоговых объектов...
}

// Параллельная обработка с группировкой по хешу ключа
async Task ProcessByKeyHash(IEnumerable<DataItem> items, int partitionCount)
{
    var partitions = new List<DataItem>[partitionCount];
    for (int i = 0; i < partitionCount; i++)
        partitions[i] = new List<DataItem>();
    
    // Распределение по партициям на основе хеша ключа
    foreach (var item in items)
    {
        int hash = Math.Abs(item.Key.GetHashCode()) % partitionCount;
        partitions[hash].Add(item);
    }
    
    // Параллельная обработка каждой партиции
    var tasks = partitions.Select(partition => 
        Task.Run(() => ProcessPartition(partition))
    );
    
    await Task.WhenAll(tasks);
}

async Task ProcessWithErrorIsolation(IEnumerable<DataItem> items)
{
    var results = new ConcurrentBag<ProcessResult>();
    
    await Parallel.ForEachAsync(items, async (item, ct) =>
    {
        try
        {
            var processed = await ProcessItemAsync(item);
            results.Add(new ProcessResult { Item = processed, Success = true });
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, $"Error processing item {item.Id}");
            results.Add(new ProcessResult { Item = item, Success = false, Error = ex.Message });
        }
    });
    
    // Анализируем результаты, возможно, повторяем обработку для не удавшихся элементов
}

async Task<T> RetryWithBackoffAsync<T>(Func<Task<T>> operation, int maxRetries = 3)
{
    int attempt = 0;
    while (true)
    {
        try
        {
            return await operation();
        }
        catch (Exception ex) when (IsTransientException(ex) && attempt < maxRetries)
        {
            attempt++;
            int delayMs = (int)Math.Pow(2, attempt) * 100; // Экспоненциальная задержка
            _logger.LogWarning($"Retry attempt {attempt} after {delayMs}ms delay");
            await Task.Delay(delayMs);
        }
    }
}
 
// Использование:
var result = await RetryWithBackoffAsync(() => _api.FetchDataAsync(id));

async Task ProcessWithTransaction(IEnumerable<DataItem> items)
{
    using var connection = new SqlConnection(_connectionString);
    await connection.OpenAsync();
    
    using var transaction = connection.BeginTransaction();
    
    try
    {
        foreach (var item in items)
        {
            // Выполняем операции в рамках транзакции
            await SaveItemAsync(connection, transaction, item);
        }
        
        // Если всё успешно - фиксируем изменения
        transaction.Commit();
    }
    catch (Exception)
    {
        // При ошибке - откатываем все изменения
        transaction.Rollback();
        throw;
    }
}

async Task ProcessWithCheckpoints(IEnumerable<DataItem> items, string checkpointFile)
{
    // Загружаем информацию о последней контрольной точке
    var lastProcessedId = await LoadCheckpointAsync(checkpointFile);
    
    // Отфильтровываем уже обработанные элементы
    var remainingItems = items.Where(i => i.Id > lastProcessedId).ToList();
    
    foreach (var item in remainingItems)
    {
        await ProcessItemAsync(item);
        
        // Сохраняем контрольную точку после каждой обработки
        // или с определенной периодичностью
        await SaveCheckpointAsync(checkpointFile, item.Id);
    }
}

int CalculateOptimalDegreeOfParallelism()
{
    // Базовая эвристика: используем количество логических процессоров
    int processors = Environment.ProcessorCount;
    
    // Для IO-bound операций можно использовать больше потоков
    if (IsIoBoundOperation())
        return processors * 2;
        
    // Для CPU-bound операций обычно оптимально число потоков = числу ядер
    return processors;
}

async Task ProcessWithDynamicPartitioning(IEnumerable<DataItem> items)
{
    // Создаем партишионер с динамическим распределением элементов
    var partitioner = Partitioner.Create(items, loadBalance: true);
    
    // Parallel.ForEach сам распределит элементы между потоками
    Parallel.ForEach(
        partitioner,
        new ParallelOptions { MaxDegreeOfParallelism = CalculateOptimalDegreeOfParallelism() },
        item => ProcessItem(item)
    );
}

class AdaptiveEtlPipeline
{
    private int _currentParallelism;
    private readonly PerformanceCounter _cpuCounter;
    private readonly PerformanceCounter _memoryCounter;
    
    public AdaptiveEtlPipeline()
    {
        _currentParallelism = Environment.ProcessorCount;
        _cpuCounter = new PerformanceCounter("Processor", "% Processor Time", "_Total");
        _memoryCounter = new PerformanceCounter("Memory", "Available MBytes");
    }
    
    public async Task ExecuteWithAdaptiveScaling(Func<int, Task> workload)
    {
        // Запускаем мониторинг в отдельном потоке
        var monitoringTask = Task.Run(async () =>
        {
            while (true)
            {
                await Task.Delay(5000); // Проверка каждые 5 секунд
                
                float cpuUsage = _cpuCounter.NextValue();
                float availableMemory = _memoryCounter.NextValue();
                
                // Адаптируем параллелизм в зависимости от нагрузки
                if (cpuUsage > 90 && _currentParallelism > 1)
                    _currentParallelism--;
                else if (cpuUsage < 70 && availableMemory > 1024)
                    _currentParallelism++;
            }
        });
        
        // Выполняем основную работу с текущей степенью параллелизма
        await workload(_currentParallelism);
        
        // Останавливаем мониторинг
        monitoringTask.Dispose();
    }
}

// Простейший профайлер времени
class SimpleProfiler : IDisposable
{
    private readonly string _operationName;
    private readonly Stopwatch _stopwatch;
    
    public SimpleProfiler(string operationName)
    {
        _operationName = operationName;
        _stopwatch = Stopwatch.StartNew();
    }
    
    public void Dispose()
    {
        _stopwatch.Stop();
        Console.WriteLine($"{_operationName}: {_stopwatch.ElapsedMilliseconds} ms");
    }
}
 
// Использование:
using (new SimpleProfiler("Extracting data"))
{
    var data = await _extractor.ExtractDataAsync(query);
}

// Вместо этого
foreach (var item in items)
{
    await connection.ExecuteAsync(insertSql, item);
}
 
// Делайте так
await connection.ExecuteAsync(batchInsertSql, items);

// Неоптимально: блокировка потока
var results = new List<Result>();
foreach (var item in items)
{
    var result = ProcessItem(item); // Синхронная операция
    results.Add(result);
}
 
// Оптимально: асинхронная обработка с потреблением по мере готовности
await foreach (var result in ProcessItemsAsync(items))
{
    // Обработка результата по мере готовности
}

// Настройка пула соединений
var connectionString = new SqlConnectionStringBuilder
{
    DataSource = "server",
    InitialCatalog = "database",
    IntegratedSecurity = true,
    // Ключевые параметры пула соединений:
    MinPoolSize = 10,
    MaxPoolSize = 100
}.ToString();

// Вместо этого
var query = "SELECT * FROM LargeTable";
 
// Делайте так
var query = "SELECT id, name, created_at FROM LargeTable WHERE updated_at > @lastSyncTime";

public class EtlMetrics
{
    private readonly Stopwatch _totalTime = new();
    private long _processedItems;
    private long _failedItems;
    
    public void StartProcess() => _totalTime.Start();
    public void StopProcess() => _totalTime.Stop();
    
    public void IncrementProcessed() => Interlocked.Increment(ref _processedItems);
    public void IncrementFailed() => Interlocked.Increment(ref _failedItems);
    
    public double GetThroughput() => _processedItems / (_totalTime.ElapsedMilliseconds / 1000.0);
    public double GetErrorRate() => (double)_failedItems / _processedItems;
    
    public void LogMetrics()
    {
        Console.WriteLine($"Processed: {_processedItems}, Failed: {_failedItems}");
        Console.WriteLine($"Total time: {_totalTime.ElapsedMilliseconds / 1000.0} s");
        Console.WriteLine($"Throughput: {GetThroughput():F2} items/s");
        Console.WriteLine($"Error rate: {GetErrorRate():P}");
    }
}

// Интеграция с Prometheus для экспорта метрик
public class PrometheusMetricsExporter
{
private readonly Counter _itemsProcessed;
private readonly Counter _itemsFailed;
private readonly Gauge _currentBatchSize;
private readonly Histogram _processingTime;
 
public PrometheusMetricsExporter()
{
    _itemsProcessed = Metrics.CreateCounter("etl_items_processed", "Total items processed");
    _itemsFailed = Metrics.CreateCounter("etl_items_failed", "Total items failed");
    _currentBatchSize = Metrics.CreateGauge("etl_current_batch_size", "Current batch size");
    _processingTime = Metrics.CreateHistogram("etl_processing_time", "Time to process batch",
        new HistogramConfiguration
        {
            Buckets = Histogram.ExponentialBuckets(start: 0.1, factor: 2, count: 10)
        });
}
 
public void RecordItemsProcessed(int count) => _itemsProcessed.Inc(count);
public void RecordItemFailed() => _itemsFailed.Inc();
public void SetCurrentBatchSize(int size) => _currentBatchSize.Set(size);
public IDisposable MeasureProcessingTime() => _processingTime.NewTimer();
}

// Инструментация использования памяти
void LogMemoryUsage(string checkpoint)
{
GC.Collect(); // Форсируем сборку мусора для получения точных данных
var memoryInMB = Process.GetCurrentProcess().WorkingSet64 / (1024 * 1024);
Console.WriteLine($"{checkpoint}: {memoryInMB} MB");
}
 
// Анализ поколений сборки мусора
void LogGCStats(string checkpoint)
{
var gen0 = GC.CollectionCount(0);
var gen1 = GC.CollectionCount(1);
var gen2 = GC.CollectionCount(2);
Console.WriteLine($"{checkpoint} - GC: Gen0={gen0}, Gen1={gen1}, Gen2={gen2}");
}

// Плохой пример - создание временных объектов в цикле
foreach (var item in largeDataSet)
{
var tempList = new List<string>();
foreach (var field in item.Fields)
{
    tempList.Add(TransformField(field));
}
ProcessTransformedItem(new TransformedItem(tempList));
}
 
// Хороший пример - переиспользование объектов
var reusableList = new List<string>(capacity: 100);
foreach (var item in largeDataSet)
{
reusableList.Clear(); // Сохраняет выделенную память
foreach (var field in item.Fields)
{
    reusableList.Add(TransformField(field));
}
ProcessTransformedItem(new TransformedItem(reusableList.ToArray()));
}

// Неоптимально - копирование коллекций
public List<TransformedData> TransformBatch(List<RawData> batch)
{
var result = new List<TransformedData>();
foreach (var item in batch)
{
    result.Add(Transform(item));
}
return result;
}
 
// Оптимально - потоковая обработка без промежуточных коллекций
public IEnumerable<TransformedData> TransformBatchStream(IEnumerable<RawData> batch)
{
foreach (var item in batch)
{
    yield return Transform(item);
}
}

// В начале приложения
// Включаем параллельную сборку мусора для многоядерных систем
GCSettings.LatencyMode = GCLatencyMode.Batch;

// Worker Pull Model - работники сами берут задачи, когда освобождаются
async Task ProcessWithWorkerPull(IEnumerable<DataItem> items, int workerCount)
{
var queue = new BlockingCollection<DataItem>(boundedCapacity: 1000);
var processingTasks = new Task[workerCount];
 
// Заполняем очередь задач
foreach (var item in items)
{
    queue.Add(item);
}
queue.CompleteAdding();
 
// Создаем рабочих, которые будут вытягивать задачи
for (int i = 0; i < workerCount; i++)
{
    processingTasks[i] = Task.Run(() =>
    {
        foreach (var item in queue.GetConsumingEnumerable())
        {
            ProcessItem(item);
        }
    });
}
 
await Task.WhenAll(processingTasks);
}

public async Task ProcessTransactionsAsync()
{
    // Создаем TPL Dataflow конвейер
    var extractBlock = new TransformBlock<DateRange, IEnumerable<Transaction>>(
        range => _repository.GetTransactionsAsync(range.From, range.To),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 }
    );
 
    var enrichBlock = new TransformBlock<IEnumerable<Transaction>, IEnumerable<EnrichedTransaction>>(
        async transactions => {
            // Параллельное обогащение транзакций
            var enrichTasks = transactions.Select(async t => {
                var userData = await _userService.GetUserDataAsync(t.UserId);
                var merchantData = await _merchantService.GetMerchantDataAsync(t.MerchantId);
                var riskScore = await _fraudDetector.CalculateRiskScoreAsync(t);
                
                return new EnrichedTransaction(t, userData, merchantData, riskScore);
            });
            
            return await Task.WhenAll(enrichTasks);
        },
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 16 }
    );
 
    var normalizeBlock = new TransformBlock<IEnumerable<EnrichedTransaction>, IEnumerable<NormalizedTransaction>>(
        transactions => transactions.AsParallel()
                                   .Select(t => _normalizer.Normalize(t))
                                   .ToList(),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 8 }
    );
 
    var loadBlock = new ActionBlock<IEnumerable<NormalizedTransaction>>(
        async batch => await _dataWarehouse.BulkInsertAsync(batch),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 2 }
    );
 
    // Соединяем блоки в конвейер
    extractBlock.LinkTo(enrichBlock, new DataflowLinkOptions { PropagateCompletion = true });
    enrichBlock.LinkTo(normalizeBlock, new DataflowLinkOptions { PropagateCompletion = true });
    normalizeBlock.LinkTo(loadBlock, new DataflowLinkOptions { PropagateCompletion = true });
 
    // Разбиваем день на часовые интервалы для параллельной обработки
    var today = DateTime.Today;
    var timeRanges = Enumerable.Range(0, 24)
                              .Select(hour => new DateRange(
                                 today.AddHours(hour), 
                                 today.AddHours(hour + 1)
                              ));
 
    // Запускаем обработку
    foreach (var range in timeRanges)
    {
        await extractBlock.SendAsync(range);
    }
    
    extractBlock.Complete();
    await loadBlock.Completion;
}