Продвинутый ETL на C# из OLTP БД в хранилище

public interface IPipelineComponent<TInput, TOutput>
{
    Task<TOutput> ProcessAsync(TInput input, CancellationToken cancellationToken);
}
 
// Пример компонента трансформации
public class CustomerDataTransformer : IPipelineComponent<List<CustomerDto>, List<CustomerWarehouseEntity>>
{
    public async Task<List<CustomerWarehouseEntity>> ProcessAsync(
        List<CustomerDto> input, 
        CancellationToken cancellationToken)
    {
        // Логика трансформации
        var result = new List<CustomerWarehouseEntity>();
        foreach (var item in input)
        {
            cancellationToken.ThrowIfCancellationRequested();
            result.Add(new CustomerWarehouseEntity
            {
                CustomerId = item.Id,
                FullName = $"{item.FirstName} {item.LastName}",
                TotalOrders = item.Orders.Count,
                AverageOrderValue = item.Orders.Any() 
                    ? item.Orders.Average(o => o.Total) 
                    : 0
            });
        }
        return result;
    }
}

public class LoggingPipelineDecorator<TInput, TOutput> : IPipelineComponent<TInput, TOutput>
{
    private readonly IPipelineComponent<TInput, TOutput> _inner;
    private readonly ILogger _logger;
 
    public LoggingPipelineDecorator(
        IPipelineComponent<TInput, TOutput> inner,
        ILogger logger)
    {
        _inner = inner;
        _logger = logger;
    }
 
    public async Task<TOutput> ProcessAsync(TInput input, CancellationToken cancellationToken)
    {
        _logger.LogInformation("Начало обработки {ComponentType}", _inner.GetType().Name);
        var stopwatch = Stopwatch.StartNew();
        
        try
        {
            var result = await _inner.ProcessAsync(input, cancellationToken);
            stopwatch.Stop();
            _logger.LogInformation("Завершено за {ElapsedMs} мс", stopwatch.ElapsedMilliseconds);
            return result;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Ошибка в компоненте {ComponentType}", _inner.GetType().Name);
            throw;
        }
    }
}

public async Task<List<TResult>> ProcessDataBatchesAsync<TSource, TResult>(
    IEnumerable<TSource> sourceData,
    Func<IEnumerable<TSource>, Task<IEnumerable<TResult>>> processorFunc,
    int batchSize = 1000,
    int maxDegreeOfParallelism = 4)
{
    var results = new ConcurrentBag<TResult>();
    var batches = sourceData.Chunk(batchSize).ToList();
    
    await Parallel.ForEachAsync(
        batches,
        new ParallelOptions { MaxDegreeOfParallelism = maxDegreeOfParallelism },
        async (batch, ct) =>
        {
            var batchResults = await processorFunc(batch);
            foreach (var result in batchResults)
            {
                results.Add(result);
            }
        });
    
    return results.ToList();
}

// Управление процессом через EF Core
using (var context = new EtlControlContext())
{
    var etlProcess = await context.EtlProcesses
        .Include(p => p.Sources)
        .Include(p => p.Targets)
        .FirstOrDefaultAsync(p => p.Name == "CustomerSync");
    
    etlProcess.LastRunStarted = DateTime.UtcNow;
    etlProcess.Status = EtlProcessStatus.Running;
    await context.SaveChangesAsync();
 
    // Массовое извлечение данных через Dapper
    using (var connection = new SqlConnection(etlProcess.Sources.First().ConnectionString))
    {
        var lastSyncPoint = etlProcess.LastSuccessfulRunCompleted ?? DateTime.MinValue;
        
        var customers = await connection.QueryAsync<CustomerDto>(@"
            SELECT c.Id, c.FirstName, c.LastName, c.Email, c.CreatedAt, c.UpdatedAt
            FROM Customers c
            WHERE c.UpdatedAt > @LastSync
            ORDER BY c.UpdatedAt",
            new { LastSync = lastSyncPoint });
        
        // Обработка данных...
        
        // Обновление статуса через EF Core
        etlProcess.LastSuccessfulRunCompleted = DateTime.UtcNow;
        etlProcess.Status = EtlProcessStatus.Completed;
        await context.SaveChangesAsync();
    }
}

public interface ISourceRepository<TEntity, TKey>
{
    Task<IEnumerable<TEntity>> GetModifiedSinceAsync(DateTime timestamp);
    Task<TEntity> GetByKeyAsync(TKey key);
    Task MarkAsSyncedAsync(TKey key, DateTime syncTimestamp);
}
 
public interface ITargetRepository<TEntity, TKey>
{
    Task UpsertAsync(IEnumerable<TEntity> entities);
    Task DeleteAsync(IEnumerable<TKey> keys);
}

public interface IUnitOfWork : IDisposable
{
    ISourceRepository<Customer, int> Customers { get; }
    ISourceRepository<Order, Guid> Orders { get; }
    ITargetRepository<CustomerWarehouseEntity, int> WarehouseCustomers { get; }
    
    Task BeginTransactionAsync();
    Task CommitAsync();
    Task RollbackAsync();
}
 
// Использование
public async Task SyncCustomersWithOrdersAsync(DateTime syncPoint)
{
    using var unitOfWork = _unitOfWorkFactory.Create();
    try
    {
        await unitOfWork.BeginTransactionAsync();
        
        var customers = await unitOfWork.Customers.GetModifiedSinceAsync(syncPoint);
        var customerIds = customers.Select(c => c.Id).ToList();
        var orders = await unitOfWork.Orders.GetByCustomerIdsAsync(customerIds);
        
        // Трансформация и объединение данных
        var warehouseEntities = MapToWarehouseEntities(customers, orders);
        
        await unitOfWork.WarehouseCustomers.UpsertAsync(warehouseEntities);
        
        await unitOfWork.CommitAsync();
    }
    catch
    {
        await unitOfWork.RollbackAsync();
        throw;
    }
}

await using (var transaction = await connection.BeginTransactionAsync())
{
    try
    {
        // Работа с транзакцией
        await transaction.CommitAsync();
    }
    catch
    {
        await transaction.RollbackAsync();
        throw;
    }
}

public async Task ProcessLargeDataSetAsync()
{
    await foreach (var batch in GetDataBatchesAsync())
    {
        await TransformAndLoadBatchAsync(batch);
        // Явно высвобождаем ссылки
        batch.Clear();
    }
}
 
private async IAsyncEnumerable<List<SourceData>> GetDataBatchesAsync(
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    using var connection = new SqlConnection(_connectionString);
    await connection.OpenAsync(cancellationToken);
    
    using var command = connection.CreateCommand();
    command.CommandText = "SELECT * FROM LargeTable WITH (NOLOCK)";
    
    using var reader = await command.ExecuteReaderAsync(CommandBehavior.SequentialAccess, cancellationToken);
    
    var batch = new List<SourceData>(1000);
    while (await reader.ReadAsync(cancellationToken))
    {
        cancellationToken.ThrowIfCancellationRequested();
        batch.Add(MapDataFromReader(reader));
        
        if (batch.Count >= 1000)
        {
            yield return batch;
            batch = new List<SourceData>(1000);
        }
    }
    
    if (batch.Count > 0)
    {
        yield return batch;
    }
}

// Конфигурация HttpClient для API-источников с контролем жизненного цикла
services.AddHttpClient("SourceApi", client =>
{
    client.BaseAddress = new Uri(Configuration["SourceApiBaseUrl"]);
    client.Timeout = TimeSpan.FromMinutes(5);
})
.SetHandlerLifetime(TimeSpan.FromMinutes(15)) // Регулярное обновление обработчиков
.ConfigurePrimaryHttpMessageHandler(() => new SocketsHttpHandler
{
    PooledConnectionLifetime = TimeSpan.FromMinutes(10),
    MaxConnectionsPerServer = 10 // Контроль параллельных соединений
});

private readonly ConditionalWeakTable<string, object> _metadataCache = new();
 
public async Task<EntityMetadata> GetEntityMetadataAsync(string entityName)
{
    if (_metadataCache.TryGetValue(entityName, out var metadata))
    {
        return (EntityMetadata)metadata;
    }
    
    var result = await LoadEntityMetadataAsync(entityName);
    _metadataCache.Add(entityName, result);
    return result;
}

private readonly Meter _meter = new("ETL.Pipeline");
private readonly Counter<long> _processedRowsCounter;
private readonly Histogram<double> _transformationDurationHistogram;
 
public DataTransformer()
{
    _processedRowsCounter = _meter.CreateCounter<long>("etl.rows.processed");
    _transformationDurationHistogram = _meter.CreateHistogram<double>("etl.transformation.duration.ms");
}
 
public async Task<IEnumerable<TTarget>> TransformBatchAsync<TSource, TTarget>(IEnumerable<TSource> sourceBatch)
{
    var stopwatch = Stopwatch.StartNew();
    var result = await PerformTransformationAsync(sourceBatch);
    stopwatch.Stop();
    
    _processedRowsCounter.Add(result.Count());
    _transformationDurationHistogram.Record(stopwatch.ElapsedMilliseconds);
    
    return result;
}

public async Task ProcessLargeFileAsync(string filePath)
{
    // Аренда буфера из пула вместо выделения нового
    byte[] buffer = ArrayPool<byte>.Shared.Rent(81920);
    try
    {
        using var fileStream = new FileStream(filePath, FileMode.Open, FileAccess.Read);
        int bytesRead;
        while ((bytesRead = await fileStream.ReadAsync(buffer, 0, buffer.Length)) > 0)
        {
            await ProcessDataChunkAsync(buffer, bytesRead);
        }
    }
    finally
    {
        // Возврат буфера в пул
        ArrayPool<byte>.Shared.Return(buffer);
    }
}

var sql = @"
SELECT Id, OrderDate, Total, CustomerId
FROM Orders WITH (NOLOCK)
WHERE OrderDate >= @From";
 
using var connection = new SqlConnection(connectionString);
 
var orders = await connection.QueryAsync<Order>(
    sql,
    new { From = DateTime.UtcNow.AddDays(-1) }
);

using var connection = new SqlConnection(connectionString);
await connection.OpenAsync();
 
using var transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted);
try
{
    var orders = await connection.QueryAsync<Order>(
        "SELECT * FROM Orders WHERE OrderDate >= @From",
        new { From = DateTime.UtcNow.AddDays(-1) },
        transaction
    );
    
    // Дальнейшая обработка
    
    await transaction.CommitAsync();
}
catch
{
    await transaction.RollbackAsync();
    throw;
}

public async IAsyncEnumerable<List<Order>> GetOrdersInBatchesAsync(
    DateTime fromDate,
    int batchSize = 10000,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    int offset = 0;
    bool hasMoreData = true;
    
    while (hasMoreData)
    {
        var sql = @"
        SELECT Id, OrderDate, Total, CustomerId
        FROM Orders WITH (NOLOCK)
        WHERE OrderDate >= @From
        ORDER BY Id
        OFFSET @Offset ROWS
        FETCH NEXT @BatchSize ROWS ONLY";
        
        using var connection = new SqlConnection(_connectionString);
        var orders = (await connection.QueryAsync<Order>(
            sql,
            new { From = fromDate, Offset = offset, BatchSize = batchSize }
        )).ToList();
        
        hasMoreData = orders.Count == batchSize;
        offset += orders.Count;
        
        if (orders.Any())
            yield return orders;
    }
}

public async Task<IEnumerable<TEntity>> GetChangesViaCdcAsync<TEntity>(
    string tableName, 
    long lastLsn)
{
    var sql = @"
    DECLARE @begin_time datetime2 = sys.fn_cdc_map_lsn_to_time(@lastLsn);
    
    SELECT * FROM cdc.fn_cdc_get_all_changes_dbo_" + tableName + @" (
        sys.fn_cdc_map_time_to_lsn('smallest greater than', @begin_time),
        sys.fn_cdc_map_time_to_lsn('largest less than or equal', GETDATE()),
        'all'
    )";
    
    using var connection = new SqlConnection(_connectionString);
    return await connection.QueryAsync<TEntity>(sql, new { lastLsn });
}

var sql = @"
SELECT * FROM Products
WHERE (ModifiedDate > @LastSyncDate)
   OR (ModifiedDate = @LastSyncDate AND Id > @LastSyncId)
ORDER BY ModifiedDate, Id
LIMIT 5000";
 
var parameters = new 
{
    LastSyncDate = _lastSync.Date,
    LastSyncId = _lastSync.Id
};

// Параллельная загрузка по хеш-сегментам
public async Task LoadDataInParallelSegmentsAsync()
{
    var tasks = new List<Task>();
    // Разбиваем по модулю ключа на 10 сегментов
    for (int i = 0; i < 10; i++)
    {
        int segmentId = i;
        tasks.Add(Task.Run(async () => {
            await LoadSegmentAsync(segmentId, 10);
        }));
    }
    await Task.WhenAll(tasks);
}
 
private async Task LoadSegmentAsync(int remainder, int modulo)
{
    var sql = @"
    SELECT * FROM LargeTable
    WHERE ABS(CAST(CHECKSUM(Id) AS bigint)) % @Modulo = @Remainder
      AND ModifiedDate > @LastSync";
    
    using var connection = new SqlConnection(_connectionString);
    var data = await connection.QueryAsync(sql, new {
        Modulo = modulo,
        Remainder = remainder,
        LastSync = _lastSyncTime
    });
    
    // Обработка сегмента данных
}

public async Task<ChangeSet> GetChangesFromLogAsync(string tableName, DateTime since)
{
    using var connection = new SqlConnection(_connectionString);
    
    // Получаем ID записей по типам изменений
    var insertedIds = await connection.QueryAsync<string>(
        "SELECT RecordId FROM ETL_ChangeLog " +
        "WHERE TableName = @Table AND ChangeType = 'I' AND ChangedAt > @Since",
        new { Table = tableName, Since = since }
    );
    
    var updatedIds = await connection.QueryAsync<string>(
        "SELECT RecordId FROM ETL_ChangeLog " +
        "WHERE TableName = @Table AND ChangeType = 'U' AND ChangedAt > @Since",
        new { Table = tableName, Since = since }
    );
    
    var deletedIds = await connection.QueryAsync<string>(
        "SELECT RecordId FROM ETL_ChangeLog " +
        "WHERE TableName = @Table AND ChangeType = 'D' AND ChangedAt > @Since",
        new { Table = tableName, Since = since }
    );
    
    // Теперь получаем актуальные данные для вставленных/обновленных записей
    var upsertRecords = new List<dynamic>();
    if (insertedIds.Any() || updatedIds.Any())
    {
        var allIds = insertedIds.Union(updatedIds).Distinct();
        var idsParam = string.Join(",", allIds.Select(id => $"'{id}'"));
        
        var records = await connection.QueryAsync(
            $"SELECT * FROM {tableName} WHERE Id IN ({idsParam})"
        );
        upsertRecords.AddRange(records);
    }
    
    return new ChangeSet
    {
        UpsertRecords = upsertRecords,
        DeletedRecordIds = deletedIds.ToList()
    };
}

public async Task<ChangeSet> DetectChangesBySnapshotComparisonAsync(
    string tableName,
    string keyColumn)
{
    // Получаем текущий снапшот
    using var connection = new SqlConnection(_connectionString);
    var currentData = await connection.QueryAsync(
        $"SELECT * FROM {tableName}"
    );
    
    // Загружаем предыдущий снапшот из нашего хранилища
    var previousData = await _snapshotStorage.GetPreviousSnapshotAsync(tableName);
    
    // Сравниваем и определяем изменения
    var currentDict = currentData.ToDictionary(
        r => ((IDictionary<string, object>)r)[keyColumn].ToString()
    );
    
    var previousDict = previousData?.ToDictionary(
        r => ((IDictionary<string, object>)r)[keyColumn].ToString()
    ) ?? new Dictionary<string, dynamic>();
    
    var upserts = currentDict.Values.Where(curr => {
        var key = ((IDictionary<string, object>)curr)[keyColumn].ToString();
        return !previousDict.ContainsKey(key) || !AreRecordsEqual(curr, previousDict[key]);
    }).ToList();
    
    var deletes = previousDict.Keys
        .Where(key => !currentDict.ContainsKey(key))
        .ToList();
    
    // Сохраняем текущий снапшот для следующего сравнения
    await _snapshotStorage.SaveSnapshotAsync(tableName, currentData);
    
    return new ChangeSet
    {
        UpsertRecords = upserts,
        DeletedRecordIds = deletes
    };
}
 
private bool AreRecordsEqual(dynamic record1, dynamic record2)
{
    var dict1 = (IDictionary<string, object>)record1;
    var dict2 = (IDictionary<string, object>)record2;
    
    return dict1.Count == dict2.Count && 
           dict1.All(kvp => dict2.ContainsKey(kvp.Key) && 
                           Equals(kvp.Value, dict2[kvp.Key]));
}

public async Task<List<string>> DetectDeletedRecordsAsync(string sourceTable, string targetTable, string keyColumn)
{
    // Получаем все ключи из источника
    using var sourceConn = new SqlConnection(_sourceConnectionString);
    var sourceKeys = (await sourceConn.QueryAsync<string>(
        $"SELECT {keyColumn} FROM {sourceTable} WITH (NOLOCK)"
    )).ToHashSet();
    
    // Получаем все ключи из целевой таблицы
    using var targetConn = new SqlConnection(_warehouseConnectionString);
    var targetKeys = (await targetConn.QueryAsync<string>(
        $"SELECT {keyColumn} FROM {targetTable}"
    )).ToHashSet();
    
    // Находим ключи, которые есть в целевой таблице, но отсутствуют в источнике
    return targetKeys.Where(k => !sourceKeys.Contains(k)).ToList();
}

public async Task ProcessDeletedRecordsInPartitionsAsync(string sourceTable, string targetTable, string keyColumn)
{
    // Разбиваем на 10 партиций по хешу ключа
    for (int partition = 0; partition < 10; partition++)
    {
        var deletedKeys = await GetDeletedKeysForPartitionAsync(
            sourceTable, targetTable, keyColumn, partition, 10);
        
        if (deletedKeys.Any())
        {
            await MarkAsDeletedInWarehouseAsync(targetTable, keyColumn, deletedKeys);
        }
    }
}

public async Task MarkRecordsAsDeletedAsync(string table, IEnumerable<string> keys)
{
    using var connection = new SqlConnection(_warehouseConnectionString);
    
    // Формируем список параметров для запроса
    var parameters = new DynamicParameters();
    parameters.Add("@DeletedDate", DateTime.UtcNow);
    
    // Формируем строку с перечислением ключей для условия IN
    var keysList = string.Join(",", keys.Select((k, i) => {
        parameters.Add($"@Key{i}", k);
        return $"@Key{i}";
    }));
    
    // Выполняем обновление
    await connection.ExecuteAsync(
        $"UPDATE {table} SET IsDeleted = 1, DeletedDate = @DeletedDate WHERE Id IN ({keysList})",
        parameters
    );
}

public class DatabaseConnectionFactory
{
    private readonly string _primaryConnectionString;
    private readonly string _replicaConnectionString;
    
    // Для операций чтения ETL используем реплику
    public SqlConnection CreateConnectionForEtlReading()
    {
        return new SqlConnection(_replicaConnectionString);
    }
    
    // Для операций записи и метаданных используем основную БД
    public SqlConnection CreateConnectionForMetadata()
    {
        return new SqlConnection(_primaryConnectionString);
    }
}

public async Task ExtractFromBackupAsync()
{
    // Восстанавливаем бэкап во временную БД
    await _databaseManager.RestoreBackupAsync(
        backupFile: "nightly_backup.bak",
        targetDatabaseName: "ETL_Temp_Extract"
    );
    
    // Теперь читаем из временной БД
    using var connection = new SqlConnection(
        "Server=.;Database=ETL_Temp_Extract;Trusted_Connection=True;"
    );
    
    var data = await connection.QueryAsync("SELECT * FROM Orders");
    
    // Обработка данных...
    
    // Удаляем временную БД после использования
    await _databaseManager.DropDatabaseAsync("ETL_Temp_Extract");
}

public async Task<bool> ShouldStartExtractProcessAsync()
{
    using var connection = new SqlConnection(_monitoringDbConnection);
    
    // Проверяем текущую нагрузку на систему
    var currentLoad = await connection.QueryFirstOrDefaultAsync<int>(
        "SELECT AVG(current_load) FROM system_monitoring WHERE sample_time > DATEADD(minute, -5, GETDATE())"
    );
    
    // Получаем типичную нагрузку для текущего времени на основе исторических данных
    var hourOfDay = DateTime.Now.Hour;
    var dayOfWeek = (int)DateTime.Now.DayOfWeek;
    
    var typicalLoad = await connection.QueryFirstOrDefaultAsync<int>(
        "SELECT AVG(average_load) FROM load_patterns WHERE hour_of_day = @Hour AND day_of_week = @Day",
        new { Hour = hourOfDay, Day = dayOfWeek }
    );
    
    // Если текущая нагрузка выше типичной более чем на 20%, откладываем процесс
    return currentLoad <= typicalLoad * 1.2;
}

public async Task<List<Order>> ExtractOrdersWithAdaptiveThrottlingAsync(DateTime since)
{
    var result = new List<Order>();
    int currentBatchSize = 5000; // Начальный размер пакета
    int offset = 0;
    bool hasMoreData = true;
    
    while (hasMoreData)
    {
        // Проверяем нагрузку на систему
        int systemLoad = await GetCurrentSystemLoadAsync();
        
        // Адаптируем размер пакета к нагрузке
        if (systemLoad > 80) currentBatchSize = 500;
        else if (systemLoad > 60) currentBatchSize = 2000;
        else currentBatchSize = 5000;
        
        // Делаем запрос с адаптированным размером пакета
        var batch = await ExtractOrdersBatchAsync(since, offset, currentBatchSize);
        result.AddRange(batch);
        
        offset += batch.Count;
        hasMoreData = batch.Count == currentBatchSize;
        
        // Делаем паузу, пропорциональную нагрузке
        int delayMs = systemLoad < 50 ? 0 : (systemLoad - 50) * 10;
        if (delayMs > 0)
            await Task.Delay(delayMs);
    }
    
    return result;
}

// Перед запуском ETL создаем временный индекс
await connection.ExecuteAsync(
    "CREATE NONCLUSTERED INDEX IX_Temp_Orders_Extract ON Orders(OrderDate) INCLUDE(CustomerId, Total)"
);
 
try
{
    // Выполняем извлечение данных
    var orders = await connection.QueryAsync<Order>(
        "SELECT Id, OrderDate, Total, CustomerId FROM Orders WITH (INDEX = IX_Temp_Orders_Extract) WHERE OrderDate >= @Since",
        new { Since = lastSyncDate }
    );
    
    // Обработка данных...
}
finally
{
    // Удаляем временный индекс
    await connection.ExecuteAsync("DROP INDEX IX_Temp_Orders_Extract ON Orders");
}

var orderItems = await connection.QueryAsync<OrderItem>(
    "SELECT * FROM OrderItems WITH (READPAST, TABLOCK) WHERE OrderId IN @OrderIds",
    new { OrderIds = orderIds }
);

// В OLTP-системе после изменения данных
public async Task EnqueueDataChangeAsync(string entityType, string entityId, ChangeType changeType)
{
    var message = new DataChangeMessage
    {
        EntityType = entityType,
        EntityId = entityId,
        ChangeType = changeType,
        Timestamp = DateTime.UtcNow
    };
    
    await _messageQueue.SendMessageAsync(JsonSerializer.Serialize(message));
}
 
// В ETL-процессе
public async Task ProcessDataChangeQueueAsync()
{
    while (true)
    {
        var message = await _messageQueue.ReceiveMessageAsync();
        if (message == null) break;
        
        var change = JsonSerializer.Deserialize<DataChangeMessage>(message);
        
        // Обработка изменения в зависимости от типа
        switch (change.EntityType)
        {
            case "Order":
                await ProcessOrderChangeAsync(change.EntityId, change.ChangeType);
                break;
            // Другие типы сущностей...
        }
    }
}

public async Task<TableSchema> DiscoverTableSchemaAsync(string tableName)
{
    var schema = new TableSchema { Name = tableName, Columns = new List<ColumnInfo>() };
    
    using var connection = new SqlConnection(_connectionString);
    
    // Получаем схему таблицы через системный каталог
    var columns = await connection.QueryAsync<dynamic>(@"
        SELECT c.name, t.name as type_name, c.max_length, c.is_nullable
        FROM sys.columns c
        JOIN sys.types t ON c.system_type_id = t.system_type_id
        JOIN sys.tables tbl ON c.object_id = tbl.object_id
        WHERE tbl.name = @TableName",
        new { TableName = tableName });
    
    foreach (var col in columns)
    {
        schema.Columns.Add(new ColumnInfo
        {
            Name = col.name,
            DataType = MapSqlTypeToClrType(col.type_name),
            IsNullable = col.is_nullable
        });
    }
    
    return schema;
}

public async Task TrackSchemaChangesAsync(string tableName, TableSchema newSchema)
{
    var existingSchema = await _schemaRepository.GetLatestSchemaAsync(tableName);
    
    if (existingSchema == null)
    {
        await _schemaRepository.SaveSchemaVersionAsync(tableName, newSchema, 1);
        return;
    }
    
    // Выявляем различия в схемах
    var differences = DetectSchemaDifferences(existingSchema, newSchema);
    if (differences.Any())
    {
        int newVersion = existingSchema.Version + 1;
        await _schemaRepository.SaveSchemaVersionAsync(tableName, newSchema, newVersion);
        
        // Регистрируем изменения для аудита
        foreach (var diff in differences)
        {
            await _schemaRepository.LogSchemaChangeAsync(
                tableName, diff.ChangeType, diff.ColumnName, diff.OldType, diff.NewType);
        }
    }
}

public async Task<IEnumerable<dynamic>> ExtractWithSchemaAdaptationAsync(
    string tableName, TableSchema schema)
{
    // Формируем список полей на основе известной схемы
    var columnNames = string.Join(", ", schema.Columns.Select(c => c.Name));
    
    var sql = $"SELECT {columnNames} FROM {tableName}";
    
    using var connection = new SqlConnection(_connectionString);
    return await connection.QueryAsync(sql);
}

public object TransformValueWithSchemaHistory(
    string columnName, object value, List<SchemaChange> columnHistory)
{
    foreach (var change in columnHistory.OrderByDescending(c => c.Version))
    {
        if (change.ChangeType == SchemaChangeType.TypeChanged)
        {
            value = ConvertType(value, change.NewType);
        }
        else if (change.ChangeType == SchemaChangeType.Renamed && 
                 change.NewName == columnName)
        {
            // Используем старое имя для поиска в источнике
            columnName = change.OldName;
        }
    }
    
    return value;
}

public class EtlSettings
{
    public List<SourceSettings> Sources { get; set; }
    public List<TransformationSettings> Transformations { get; set; }
    public List<TargetSettings> Targets { get; set; }
}
 
public class SourceSettings
{
    public string Name { get; set; }
    public string ConnectionString { get; set; }
    public List<EntitySettings> Entities { get; set; }
}
 
// Другие классы настроек опущены для краткости

public void ConfigureServices(IServiceCollection services, IConfiguration configuration)
{
    // Привязываем конфигурацию к классам настроек
    services.Configure<EtlSettings>(configuration.GetSection("EtlSettings"));
    
    // Регистрируем базовые сервисы
    services.AddSingleton<IEtlProcessManager, EtlProcessManager>();
    services.AddSingleton<IConnectionFactory, ConnectionFactory>();
    services.AddSingleton<IMetadataRepository, MetadataRepository>();
    
    // Регистрируем трансформеры из конфигурации
    var transformers = configuration.GetSection("EtlSettings:Transformations")
        .Get<List<TransformationSettings>>();
    
    foreach (var transformer in transformers)
    {
        // Динамически создаем тип трансформера из строки в конфигурации
        var transformerType = Type.GetType(transformer.TransformerType);
        
        // Находим интерфейс трансформера
        var interfaceType = transformerType.GetInterfaces()
            .First(i => i.IsGenericType && i.GetGenericTypeDefinition() == typeof(ITransformer<,>));
        
        // Регистрируем трансформер в DI
        services.AddTransient(interfaceType, transformerType);
    }
}

public class OrderProcessor
{
    private readonly IOptions<EtlSettings> _settings;
    private readonly IConnectionFactory _connectionFactory;
    private readonly ITransformer<OrderDto, WarehouseOrderFact> _transformer;
    
    public OrderProcessor(
        IOptions<EtlSettings> settings,
        IConnectionFactory connectionFactory,
        ITransformer<OrderDto, WarehouseOrderFact> transformer)
    {
        _settings = settings;
        _connectionFactory = connectionFactory;
        _transformer = transformer;
    }
    
    public async Task ProcessAsync()
    {
        // Получаем настройки из конфигурации
        var orderSettings = _settings.Value.Sources
            .SelectMany(s => s.Entities)
            .First(e => e.Name == "Orders");
            
        // Создаем подключение через фабрику
        using var connection = _connectionFactory.Create(
            _settings.Value.Sources.First(s => s.Name == "SalesDatabase").ConnectionString);
            
        // Остальная логика...
    }
}

public interface IPipelineStep<in TInput, out TOutput>
{
    TOutput Process(TInput input);
}
 
public interface IAsyncPipelineStep<in TInput, out TOutput>
{
    Task<TOutput> ProcessAsync(TInput input, CancellationToken cancellationToken = default);
}

public class Pipeline<TInput, TOutput>
{
    private readonly List<Func<object, Task<object>>> _steps = new();
 
    public Pipeline<TInput, TIntermediate> AddStep<TIntermediate>(
        IAsyncPipelineStep<TInput, TIntermediate> step)
    {
        _steps.Add(async input => await step.ProcessAsync((TInput)input));
        return new Pipeline<TInput, TIntermediate>(_steps);
    }
 
    public async Task<TOutput> ExecuteAsync(TInput input)
    {
        object current = input;
        foreach (var step in _steps)
        {
            current = await step(current);
        }
        return (TOutput)current;
    }
 
    // Конструктор и другие методы опущены для краткости
}

public Pipeline<TInput, (TOutput1, TOutput2)> AddParallelSteps<TOutput1, TOutput2>(
    IAsyncPipelineStep<TInput, TOutput1> step1,
    IAsyncPipelineStep<TInput, TOutput2> step2)
{
    _steps.Add(async input =>
    {
        var input1 = (TInput)input;
        var task1 = step1.ProcessAsync(input1);
        var task2 = step2.ProcessAsync(input1);
        await Task.WhenAll(task1, task2);
        return (await task1, await task2);
    });
    
    return new Pipeline<TInput, (TOutput1, TOutput2)>(_steps);
}

public class RetryPipelineDecorator<TInput, TOutput> : IAsyncPipelineStep<TInput, TOutput>
{
    private readonly IAsyncPipelineStep<TInput, TOutput> _inner;
    private readonly int _maxRetries;
    private readonly TimeSpan _delay;
 
    public RetryPipelineDecorator(
        IAsyncPipelineStep<TInput, TOutput> inner,
        int maxRetries = 3,
        TimeSpan? delay = null)
    {
        _inner = inner;
        _maxRetries = maxRetries;
        _delay = delay ?? TimeSpan.FromSeconds(2);
    }
 
    public async Task<TOutput> ProcessAsync(TInput input, CancellationToken cancellationToken)
    {
        Exception lastException = null;
        
        for (int attempt = 0; attempt <= _maxRetries; attempt++)
        {
            try
            {
                return await _inner.ProcessAsync(input, cancellationToken);
            }
            catch (Exception ex) when (attempt < _maxRetries)
            {
                lastException = ex;
                await Task.Delay(_delay, cancellationToken);
            }
        }
        
        throw new AggregateException($"Failed after {_maxRetries} retries", lastException);
    }
}

var pipeline = new Pipeline<List<OrderDto>, List<WarehouseOrderFact>>()
    .AddStep(new OrdersDeduplicator())
    .AddStep(new OrdersValidator())
    .AddStep(new OrdersTransformer())
    .AddStep(new OrdersEnricher(_customerRepository))
    .AddStep(new RetryPipelineDecorator<List<WarehouseOrderFact>, List<WarehouseOrderFact>>(
        new WarehouseOrderWriter(_warehouseConnection)));
 
var result = await pipeline.ExecuteAsync(sourceOrders);

public record Order(
    string Id,
    DateTime OrderDate,
    decimal Total,
    string CustomerId
);
 
public record OrderFact(
    string OrderId,
    string OrderDateKey,
    decimal TotalAmount,
    string CustomerKey
);
 
// Трансформация с простым маппингом
var orderFacts = orders.Select(o => new OrderFact(
    o.Id,
    o.OrderDate.ToString("yyyyMMdd"),
    o.Total,
    o.CustomerId
)).ToList();

// Промежуточная модель с бизнес-логикой
public class OrderTransformationContext
{
    public Order SourceOrder { get; }
    public Customer Customer { get; }
    public List<OrderLine> OrderLines { get; }
    
    // Расчетные поля для бизнес-логики
    public decimal DiscountPercentage => CalculateDiscountPercentage();
    public bool IsHighPriority => Customer.Tier == "Premium" || TotalWithTax > 10000;
    public decimal TotalWithTax => SourceOrder.Total * (1 + GetTaxRate());
    
    private decimal CalculateDiscountPercentage() 
    {
        // Сложная логика определения скидки
        if (Customer.LoyaltySince < DateTime.Now.AddYears(-5))
            return 0.15m;
        if (OrderLines.Count > 10)
            return 0.1m;
        return 0.05m;
    }
    
    private decimal GetTaxRate() 
    {
        // Логика определения налоговой ставки по региону
        return Customer.Region switch 
        {
            "EU" => 0.21m,
            "US" => 0.08m,
            _ => 0.18m
        };
    }
}

public class OrderValidator : IPipelineStep<List<Order>, ValidationResult<List<Order>>>
{
    public ValidationResult<List<Order>> Process(List<Order> input)
    {
        var validator = new OrderRuleValidator();
        var errors = new List<ValidationError>();
        
        foreach (var order in input)
        {
            var result = validator.Validate(order);
            if (!result.IsValid)
            {
                errors.AddRange(result.Errors.Select(e => 
                    new ValidationError(order.Id, e.PropertyName, e.ErrorMessage)));
            }
        }
        
        return new ValidationResult<List<Order>>(input, errors);
    }
}
 
public class OrderRuleValidator : AbstractValidator<Order>
{
    public OrderRuleValidator()
    {
        RuleFor(o => o.Id).NotEmpty();
        RuleFor(o => o.OrderDate).NotEmpty().Must(d => d <= DateTime.Now)
            .WithMessage("Дата заказа не может быть в будущем");
        RuleFor(o => o.Total).GreaterThan(0)
            .WithMessage("Сумма заказа должна быть положительной");
        RuleFor(o => o.CustomerId).NotEmpty();
    }
}

public class DataCleaner : IPipelineStep<List<Order>, List<Order>>
{
    public List<Order> Process(List<Order> input)
    {
        return input.Select(o => new Order(
            o.Id?.Trim(),
            o.OrderDate,
            Math.Round(o.Total, 2),  // Округляем до копеек
            o.CustomerId?.Trim().ToUpper()  // Нормализуем ID клиента
        )).ToList();
    }
}

public class DuplicateRemover<T, TKey> : IPipelineStep<List<T>, List<T>>
{
    private readonly Func<T, TKey> _keySelector;
    private readonly IEqualityComparer<TKey> _comparer;
    
    public DuplicateRemover(
        Func<T, TKey> keySelector,
        IEqualityComparer<TKey> comparer = null)
    {
        _keySelector = keySelector;
        _comparer = comparer ?? EqualityComparer<TKey>.Default;
    }
    
    public List<T> Process(List<T> input)
    {
        return input
            .GroupBy(_keySelector, _comparer)
            .Select(g => g.First())
            .ToList();
    }
}
 
// Использование
var uniqueOrders = new DuplicateRemover<Order, string>(o => o.Id)
    .Process(orders);

public class ConflictResolver<T, TKey> : IPipelineStep<List<T>, List<T>>
{
    private readonly Func<T, TKey> _keySelector;
    private readonly Func<IEnumerable<T>, T> _conflictResolver;
    
    public ConflictResolver(
        Func<T, TKey> keySelector,
        Func<IEnumerable<T>, T> conflictResolver)
    {
        _keySelector = keySelector;
        _conflictResolver = conflictResolver;
    }
    
    public List<T> Process(List<T> input)
    {
        return input
            .GroupBy(_keySelector)
            .Select(g => g.Count() > 1 ? _conflictResolver(g) : g.First())
            .ToList();
    }
}
 
// Выбираем версию с наиболее поздней датой модификации
var resolvedOrders = new ConflictResolver<Order, string>(
    o => o.Id,
    conflicts => conflicts.OrderByDescending(o => o.ModifiedDate).First())
    .Process(allOrders);

public class ParallelTransformer<TSource, TTarget> : IPipelineStep<List<TSource>, List<TTarget>>
{
  private readonly Func<TSource, TTarget> _transformFunc;
  private readonly int _degreeOfParallelism;
  
  public ParallelTransformer(
      Func<TSource, TTarget> transformFunc,
      int degreeOfParallelism = 0)  // 0 означает автоматический выбор
  {
      _transformFunc = transformFunc;
      _degreeOfParallelism = degreeOfParallelism;
  }
  
  public List<TTarget> Process(List<TSource> input)
  {
      return input
          .AsParallel()
          .WithDegreeOfParallelism(_degreeOfParallelism > 0 
              ? _degreeOfParallelism 
              : Environment.ProcessorCount)
          .Select(_transformFunc)
          .ToList();
  }
}

public class ParallelAsyncTransformer<TSource, TTarget> : IAsyncPipelineStep<List<TSource>, List<TTarget>>
{
  private readonly Func<TSource, Task<TTarget>> _transformFunc;
  private readonly int _maxConcurrency;
  
  public ParallelAsyncTransformer(
      Func<TSource, Task<TTarget>> transformFunc,
      int maxConcurrency = 10)
  {
      _transformFunc = transformFunc;
      _maxConcurrency = maxConcurrency;
  }
  
  public async Task<List<TTarget>> ProcessAsync(
      List<TSource> input, 
      CancellationToken cancellationToken)
  {
      var semaphore = new SemaphoreSlim(_maxConcurrency);
      var tasks = input.Select(async item => 
      {
          try
          {
              await semaphore.WaitAsync(cancellationToken);
              return await _transformFunc(item);
          }
          finally
          {
              semaphore.Release();
          }
      });
      
      return (await Task.WhenAll(tasks)).ToList();
  }
}

public class EcommerceOrderValidator : AbstractValidator<Order>
{
  private readonly IProductRepository _productRepo;
  private readonly ICustomerRepository _customerRepo;
  
  public EcommerceOrderValidator(
      IProductRepository productRepo,
      ICustomerRepository customerRepo)
  {
      _productRepo = productRepo;
      _customerRepo = customerRepo;
      
      // Структурная валидация
      RuleFor(o => o.Id).NotEmpty();
      RuleFor(o => o.OrderDate).NotEmpty();
      
      // Бизнес-правила уровня записи
      RuleFor(o => o.ShippingAddress)
          .NotEmpty()
          .When(o => o.DeliveryType != "Digital");
          
      RuleFor(o => o.Total)
          .GreaterThanOrEqualTo(0)
          .LessThan(50000)  // Подозрительно большие заказы выделяем для проверки
          .WithMessage("Заказ на сумму свыше 50,000 требует проверки");
          
      // Кросс-проверки с другими системами
      RuleFor(o => o.CustomerId)
          .MustAsync(async (id, ct) => await _customerRepo.ExistsAsync(id, ct))
          .WithMessage("Указанный клиент не существует");
          
      RuleFor(o => o)
          .MustAsync(async (order, ct) => await ValidateProductAvailabilityAsync(order, ct))
          .WithMessage("Один или несколько товаров недоступны");
  }
  
  private async Task<bool> ValidateProductAvailabilityAsync(Order order, CancellationToken ct)
  {
      foreach (var line in order.Lines)
      {
          var product = await _productRepo.GetByIdAsync(line.ProductId, ct);
          if (product == null || product.StockQuantity < line.Quantity)
              return false;
      }
      return true;
  }
}

public class OrderFraudDetectionValidator : AbstractValidator<Order>
{
  private readonly IOrderHistoryService _historyService;
  
  public OrderFraudDetectionValidator(IOrderHistoryService historyService)
  {
      _historyService = historyService;
      
      RuleFor(o => o)
          .MustAsync(CheckSuspiciousPatternAsync)
          .WithMessage("Обнаружен подозрительный паттерн заказов");
  }
  
  private async Task<bool> CheckSuspiciousPatternAsync(Order order, CancellationToken ct)
  {
      // Анализируем историю заказов на подозрительные паттерны
      var recentOrders = await _historyService.GetRecentOrdersAsync(
          order.CustomerId, 
          TimeSpan.FromDays(1), 
          ct);
          
      // Много мелких заказов за короткий период - признак фрода
      if (recentOrders.Count > 10 && recentOrders.All(o => o.Total < 100))
          return false;
          
      // Резкое изменение географии заказов
      var distinctCountries = recentOrders
          .Select(o => o.ShippingAddress?.Country)
          .Where(c => c != null)
          .Distinct()
          .Count();
          
      if (distinctCountries > 3)
          return false;
          
      return true;
  }
}

public class PerformanceLoggingDecorator<TInput, TOutput> : IPipelineStep<TInput, TOutput>
{
  private readonly IPipelineStep<TInput, TOutput> _inner;
  private readonly ILogger _logger;
  private readonly string _stepName;
  
  public PerformanceLoggingDecorator(
      IPipelineStep<TInput, TOutput> inner,
      ILogger logger,
      string stepName = null)
  {
      _inner = inner;
      _logger = logger;
      _stepName = stepName ?? inner.GetType().Name;
  }
  
  public TOutput Process(TInput input)
  {
      var stopwatch = Stopwatch.StartNew();
      var inputCount = CountItems(input);
      
      try
      {
          var result = _inner.Process(input);
          
          stopwatch.Stop();
          var outputCount = CountItems(result);
          
          _logger.LogInformation(
              "Шаг {StepName} обработал {InputCount} записей в {OutputCount} за {ElapsedMs}мс " +
              "({RecordsPerSecond} записей/сек)",
              _stepName, 
              inputCount, 
              outputCount,
              stopwatch.ElapsedMilliseconds,
              inputCount * 1000 / (stopwatch.ElapsedMilliseconds > 0 ? stopwatch.ElapsedMilliseconds : 1));
              
          return result;
      }
      catch (Exception ex)
      {
          stopwatch.Stop();
          _logger.LogError(ex, 
              "Ошибка в шаге {StepName} после {ElapsedMs}мс обработки",
              _stepName, 
              stopwatch.ElapsedMilliseconds);
          throw;
      }
  }
  
  private int CountItems<T>(T item)
  {
      if (item is ICollection collection)
          return collection.Count;
      if (item is IEnumerable enumerable)
          return enumerable.Cast<object>().Count();
      return 1;
  }
}

public async Task BulkInsertOrdersAsync(List<OrderFact> orders)
{
    // Преобразуем коллекцию объектов в DataTable
    var dataTable = new DataTable();
    dataTable.Columns.Add("OrderId", typeof(string));
    dataTable.Columns.Add("OrderDateKey", typeof(string));
    dataTable.Columns.Add("TotalAmount", typeof(decimal));
    dataTable.Columns.Add("CustomerKey", typeof(string));
    
    foreach (var order in orders)
    {
        dataTable.Rows.Add(
            order.OrderId,
            order.OrderDateKey,
            order.TotalAmount,
            order.CustomerKey
        );
    }
    
    using var connection = new SqlConnection(_warehouseConnectionString);
    await connection.OpenAsync();
    
    using var bulkCopy = new SqlBulkCopy(connection)
    {
        DestinationTableName = "dbo.FactOrders",
        BatchSize = 10000,
        BulkCopyTimeout = 600  // 10 минут
    };
    
    // Настраиваем соответствие колонок
    bulkCopy.ColumnMappings.Add("OrderId", "OrderId");
    bulkCopy.ColumnMappings.Add("OrderDateKey", "OrderDateKey");
    bulkCopy.ColumnMappings.Add("TotalAmount", "TotalAmount");
    bulkCopy.ColumnMappings.Add("CustomerKey", "CustomerKey");
    
    await bulkCopy.WriteToServerAsync(dataTable);
}

public async Task FastBulkInsertOrdersAsync(List<OrderFact> orders)
{
    using var connection = new SqlConnection(_warehouseConnectionString);
    await connection.OpenAsync();
    
    using var bulkCopy = new SqlBulkCopy(connection)
    {
        DestinationTableName = "dbo.FactOrders",
        BatchSize = 10000
    };
    
    // Настраиваем соответствие колонок
    bulkCopy.ColumnMappings.Add("OrderId", "OrderId");
    bulkCopy.ColumnMappings.Add("OrderDateKey", "OrderDateKey");
    bulkCopy.ColumnMappings.Add("TotalAmount", "TotalAmount");
    bulkCopy.ColumnMappings.Add("CustomerKey", "CustomerKey");
    
    // Используем FastMember для более быстрого доступа к свойствам объектов
    using var reader = ObjectReader.Create(orders, 
        "OrderId", "OrderDateKey", "TotalAmount", "CustomerKey");
    
    await bulkCopy.WriteToServerAsync(reader);
}

public async Task BulkUpsertOrdersAsync(List<OrderFact> orders)
{
    using var connection = new SqlConnection(_warehouseConnectionString);
    await connection.OpenAsync();
    
    // Создаем временную таблицу с той же структурой, что и целевая
    await connection.ExecuteAsync(@"
        CREATE TABLE #TempOrders (
            OrderId NVARCHAR(50) PRIMARY KEY,
            OrderDateKey NVARCHAR(8),
            TotalAmount DECIMAL(18,2),
            CustomerKey NVARCHAR(50)
        )");
    
    // Загружаем данные во временную таблицу
    using (var bulkCopy = new SqlBulkCopy(connection)
    {
        DestinationTableName = "#TempOrders",
        BatchSize = 10000
    })
    {
        bulkCopy.ColumnMappings.Add("OrderId", "OrderId");
        bulkCopy.ColumnMappings.Add("OrderDateKey", "OrderDateKey");
        bulkCopy.ColumnMappings.Add("TotalAmount", "TotalAmount");
        bulkCopy.ColumnMappings.Add("CustomerKey", "CustomerKey");
        
        using var reader = ObjectReader.Create(orders, 
            "OrderId", "OrderDateKey", "TotalAmount", "CustomerKey");
        
        await bulkCopy.WriteToServerAsync(reader);
    }
    
    // Выполняем MERGE для обновления/вставки данных
    await connection.ExecuteAsync(@"
        MERGE dbo.FactOrders AS target
        USING #TempOrders AS source
        ON target.OrderId = source.OrderId
        WHEN MATCHED THEN
            UPDATE SET 
                target.OrderDateKey = source.OrderDateKey,
                target.TotalAmount = source.TotalAmount,
                target.CustomerKey = source.CustomerKey
        WHEN NOT MATCHED THEN
            INSERT (OrderId, OrderDateKey, TotalAmount, CustomerKey)
            VALUES (source.OrderId, source.OrderDateKey, source.TotalAmount, source.CustomerKey);
            
        DROP TABLE #TempOrders;
    ");
}

public async Task TransactionalLoadAsync(List<OrderFact> orders, List<CustomerDim> customers)
{
    using var connection = new SqlConnection(_warehouseConnectionString);
    await connection.OpenAsync();
    
    using var transaction = await connection.BeginTransactionAsync();
    
    try
    {
        // Загрузка измерений
        using (var bulkCopy = new SqlBulkCopy(connection, SqlBulkCopyOptions.Default, transaction))
        {
            bulkCopy.DestinationTableName = "dbo.DimCustomer";
            // Настройка колонок и загрузка...
            await bulkCopy.WriteToServerAsync(CreateCustomerDataTable(customers));
        }
        
        // Загрузка фактов
        using (var bulkCopy = new SqlBulkCopy(connection, SqlBulkCopyOptions.Default, transaction))
        {
            bulkCopy.DestinationTableName = "dbo.FactOrders";
            // Настройка колонок и загрузка...
            await bulkCopy.WriteToServerAsync(CreateOrderDataTable(orders));
        }
        
        // Фиксируем транзакцию только если все успешно
        await transaction.CommitAsync();
    }
    catch
    {
        await transaction.RollbackAsync();
        throw;
    }
}

public async Task TwoPhaseLoadAsync(List<OrderFact> orders)
{
    using var connection = new SqlConnection(_warehouseConnectionString);
    await connection.OpenAsync();
    
    // Шаг 1: Загрузка во временную таблицу
    await connection.ExecuteAsync("TRUNCATE TABLE dbo.FactOrders_Staging");
    
    using (var bulkCopy = new SqlBulkCopy(connection))
    {
        bulkCopy.DestinationTableName = "dbo.FactOrders_Staging";
        // Настройка колонок и загрузка...
        await bulkCopy.WriteToServerAsync(CreateOrderDataTable(orders));
    }
    
    // Шаг 2: Атомарный своп таблиц
    await connection.ExecuteAsync(@"
        BEGIN TRANSACTION;
        
        -- Переименовываем таблицы
        EXEC sp_rename 'dbo.FactOrders', 'FactOrders_Old';
        EXEC sp_rename 'dbo.FactOrders_Staging', 'FactOrders';
        
        -- Перестраиваем индексы на новой таблице
        ALTER INDEX ALL ON dbo.FactOrders REBUILD;
        
        -- Очищаем старую таблицу
        TRUNCATE TABLE dbo.FactOrders_Old;
        EXEC sp_rename 'dbo.FactOrders_Old', 'FactOrders_Staging';
        
        COMMIT;
    ");
}

public async Task LoadWithCheckpointAsync(EtlBatch batch)
{
    // Сохраняем информацию о текущем состоянии процесса
    var checkpoint = new EtlCheckpoint
    {
        BatchId = batch.Id,
        StartTime = DateTime.UtcNow,
        State = EtlState.Loading,
        SourceQuery = batch.SourceQuery,
        RecordCount = batch.Data.Count
    };
    
    await _checkpointRepository.SaveCheckpointAsync(checkpoint);
    
    try
    {
        // Выполняем загрузку
        await BulkLoadDataAsync(batch.Data);
        
        // Обновляем статус чекпоинта
        checkpoint.EndTime = DateTime.UtcNow;
        checkpoint.State = EtlState.Completed;
        await _checkpointRepository.UpdateCheckpointAsync(checkpoint);
    }
    catch (Exception ex)
    {
        // Обновляем статус с информацией об ошибке
        checkpoint.EndTime = DateTime.UtcNow;
        checkpoint.State = EtlState.Failed;
        checkpoint.ErrorMessage = ex.Message;
        await _checkpointRepository.UpdateCheckpointAsync(checkpoint);
        
        // Запускаем процедуру восстановления
        await RecoverFromFailedLoadAsync(batch, checkpoint);
        
        throw;
    }
}

private async Task RecoverFromFailedLoadAsync(EtlBatch batch, EtlCheckpoint checkpoint)
{
    // Для временных ошибок (сетевые проблемы и т.д.) можно просто повторить
    if (IsTransientError(checkpoint.ErrorMessage))
    {
        await _logger.LogWarningAsync($"Обнаружена временная ошибка, повторная попытка загрузки для пакета {batch.Id}");
        await RetryLoadAsync(batch);
        return;
    }
    
    // Для ошибок данных может потребоваться перевалидация
    if (IsDataError(checkpoint.ErrorMessage))
    {
        await _logger.LogWarningAsync($"Обнаружена ошибка в данных для пакета {batch.Id}, запуск углубленной валидации");
        var validationResult = await DeepValidateDataAsync(batch.Data);
        if (validationResult.IsValid)
        {
            await RetryLoadAsync(batch);
        }
        else
        {
            await _logger.LogErrorAsync($"Валидация не прошла: {validationResult.ErrorMessage}");
            // Сохраняем проблемные данные для ручного анализа
            await SaveProblemDataForAnalysisAsync(batch, validationResult);
        }
        return;
    }
    
    // Для критических ошибок может потребоваться откат предыдущих успешных загрузок
    if (IsCriticalError(checkpoint.ErrorMessage))
    {
        await _logger.LogErrorAsync($"Критическая ошибка в пакете {batch.Id}, требуется откат");
        await RollbackToLastGoodStateAsync(checkpoint.BatchId);
    }
}

public async Task LoadToPartitionedTableAsync(List<FactSales> salesData, string partitionKey)
{
  // Определяем границы партиции
  var partitionRange = DeterminePartitionRange(partitionKey);
  
  using var connection = new SqlConnection(_warehouseConnectionString);
  await connection.OpenAsync();
  
  // Создаем временную таблицу для партиции
  await connection.ExecuteAsync($@"
      CREATE TABLE dbo.FactSales_Temp_{partitionKey} (
          SalesId BIGINT NOT NULL,
          DateKey INT NOT NULL,
          ProductKey INT NOT NULL,
          StoreKey INT NOT NULL,
          Quantity INT NOT NULL,
          Amount DECIMAL(18,2) NOT NULL
      );
  ");
  
  // Загружаем данные во временную таблицу
  using (var bulkCopy = new SqlBulkCopy(connection))
  {
      bulkCopy.DestinationTableName = $"dbo.FactSales_Temp_{partitionKey}";
      // Настройка маппингов...
      await bulkCopy.WriteToServerAsync(CreateDataTable(salesData));
  }
  
  // Переключаем партицию (SQL Server 2016+)
  await connection.ExecuteAsync($@"
      ALTER TABLE dbo.FactSales SWITCH PARTITION {partitionRange.PartitionNumber} 
      TO dbo.FactSales_Temp_{partitionKey} PARTITION 1;
      
      TRUNCATE TABLE dbo.FactSales_Temp_{partitionKey};
      
      ALTER TABLE dbo.FactSales_Temp_{partitionKey} SWITCH PARTITION 1
      TO dbo.FactSales PARTITION {partitionRange.PartitionNumber};
      
      DROP TABLE dbo.FactSales_Temp_{partitionKey};
  ");
}

public async Task DisableIndexesForBulkLoadAsync(string tableName)
{
  using var connection = new SqlConnection(_warehouseConnectionString);
  
  // Получаем список некластерных индексов таблицы
  var indexes = await connection.QueryAsync<string>(@"
      SELECT i.name 
      FROM sys.indexes i
      JOIN sys.tables t ON i.object_id = t.object_id
      WHERE t.name = @TableName 
        AND i.type_desc = 'NONCLUSTERED'
        AND i.is_disabled = 0",
      new { TableName = tableName });
  
  foreach (var indexName in indexes)
  {
      await connection.ExecuteAsync($"ALTER INDEX {indexName} ON {tableName} DISABLE");
  }
}
 
public async Task RebuildIndexesAfterLoadAsync(string tableName)
{
  using var connection = new SqlConnection(_warehouseConnectionString);
  
  // Получаем список отключенных индексов
  var indexes = await connection.QueryAsync<string>(@"
      SELECT i.name 
      FROM sys.indexes i
      JOIN sys.tables t ON i.object_id = t.object_id
      WHERE t.name = @TableName 
        AND i.is_disabled = 1",
      new { TableName = tableName });
  
  foreach (var indexName in indexes)
  {
      await connection.ExecuteAsync($"ALTER INDEX {indexName} ON {tableName} REBUILD");
  }
}

public async Task LoadToColumnStoreViaBlob(List<FactSales> salesData)
{
  // Шаг 1: Сохраняем данные в CSV
  var csvFilePath = Path.GetTempFileName();
  await WriteToCsvAsync(salesData, csvFilePath);
  
  // Шаг 2: Загружаем CSV в Azure Blob Storage
  var blobName = $"stage/sales/{DateTime.UtcNow:yyyyMMdd}/{Guid.NewGuid()}.csv";
  await UploadToBlobStorageAsync(csvFilePath, blobName);
  
  // Шаг 3: Используем COPY в Synapse для загрузки
  using var connection = new SqlConnection(_synapseConnectionString);
  await connection.ExecuteAsync($@"
      COPY INTO dbo.FactSales
      FROM 'https://mystorageaccount.blob.core.windows.net/{blobName}'
      WITH (
          FILE_TYPE = 'CSV',
          FIELDTERMINATOR = ',',
          FIRSTROW = 2
      )
  ");
  
  // Очистка
  File.Delete(csvFilePath);
}

public List<FactSales> OptimizeForColumnStore(List<FactSales> salesData)
{
  // Сортируем по колонкам с высокой кардинальностью
  return salesData
      .OrderBy(s => s.DateKey)
      .ThenBy(s => s.StoreKey)
      .ThenBy(s => s.ProductKey)
      .ToList();
}

public class EtlMonitoringService
{
  private readonly Meter _meter;
  private readonly Counter<long> _recordsProcessedCounter;
  private readonly Histogram<double> _processingTimeHistogram;
  private readonly ILogger _logger;
  
  public EtlMonitoringService(ILogger logger, string processName)
  {
      _logger = logger;
      _meter = new Meter($"ETL.{processName}");
      _recordsProcessedCounter = _meter.CreateCounter<long>("records.processed");
      _processingTimeHistogram = _meter.CreateHistogram<double>("processing.time.ms");
  }
  
  public void RecordBatchProcessed(int recordCount, TimeSpan duration)
  {
      _recordsProcessedCounter.Add(recordCount);
      _processingTimeHistogram.Record(duration.TotalMilliseconds);
      
      _logger.LogInformation("Обработано {RecordCount} записей за {Duration}мс", 
          recordCount, duration.TotalMilliseconds);
  }
  
  public void RecordError(Exception ex, string operation)
  {
      _logger.LogError(ex, "Ошибка при выполнении {Operation}", operation);
      
      // Отправляем оповещение для критических ошибок
      if (IsCriticalError(ex))
      {
          SendAlert($"Критическая ошибка в ETL: {ex.Message}");
      }
  }
}

public async Task<bool> DetectPerformanceAnomalyAsync(string processId, double currentThroughput)
{
  // Получаем исторические данные о производительности
  var historicalData = await _metricRepository.GetHistoricalThroughputAsync(
      processId, 
      DateTime.UtcNow.AddDays(-30), 
      DateTime.UtcNow);
      
  // Вычисляем статистические характеристики
  var mean = historicalData.Average();
  var stdDev = CalculateStandardDeviation(historicalData, mean);
  
  // Если текущая производительность отклоняется более чем на 2 стандартных отклонения,
  // считаем это аномалией
  return Math.Abs(currentThroughput - mean) > (2 * stdDev);
}

public async Task MonitorResourceUsageAsync(CancellationToken cancellationToken)
{
  var process = Process.GetCurrentProcess();
  var lastCpuTime = process.TotalProcessorTime;
  var lastMeasurementTime = DateTime.UtcNow;
  
  while (!cancellationToken.IsCancellationRequested)
  {
      await Task.Delay(TimeSpan.FromMinutes(1), cancellationToken);
      
      var currentCpuTime = process.TotalProcessorTime;
      var currentTime = DateTime.UtcNow;
      var cpuUsagePercent = (currentCpuTime - lastCpuTime).TotalMilliseconds / 
                           (Environment.ProcessorCount * currentTime.Subtract(lastMeasurementTime).TotalMilliseconds) * 100;
      
      var memoryUsageMb = process.WorkingSet64 / (1024 * 1024);
      
      _logger.LogInformation("Использование ресурсов: CPU {CpuUsage}%, Память {MemoryUsage}MB", 
          cpuUsagePercent, memoryUsageMb);
          
      if (cpuUsagePercent > 80 || memoryUsageMb > _criticalMemoryThresholdMb)
      {
          _logger.LogWarning("Превышен порог использования ресурсов!");
          SendResourceAlert(cpuUsagePercent, memoryUsageMb);
      }
      
      lastCpuTime = currentCpuTime;
      lastMeasurementTime = currentTime;
  }
}

public async Task GenerateDashboardAsync()
{
  var processes = await _etlRepository.GetAllProcessesAsync();
  var builder = new StringBuilder();
  
  builder.AppendLine("<!DOCTYPE html><html><head><title>ETL Monitor</title>");
  builder.AppendLine("<style>table { border-collapse: collapse; width: 100%; }");
  builder.AppendLine("th, td { padding: 8px; text-align: left; border: 1px solid #ddd; }");
  builder.AppendLine(".success { background-color: #dff0d8; }");
  builder.AppendLine(".failure { background-color: #f2dede; }");
  builder.AppendLine(".warning { background-color: #fcf8e3; }</style></head><body>");
  
  builder.AppendLine("<h1>ETL Monitor Dashboard</h1>");
  builder.AppendLine("<table><tr><th>Process</th><th>Last Run</th><th>Status</th><th>Duration</th><th>Records</th></tr>");
  
  foreach (var process in processes)
  {
      var statusClass = process.LastRunStatus switch
      {
          "Completed" => "success",
          "Failed" => "failure",
          "Running" => "warning",
          _ => ""
      };
      
      builder.AppendLine($"<tr class='{statusClass}'>");
      builder.AppendLine($"<td>{process.Name}</td>");
      builder.AppendLine($"<td>{process.LastRunTime}</td>");
      builder.AppendLine($"<td>{process.LastRunStatus}</td>");
      builder.AppendLine($"<td>{process.LastRunDuration}</td>");
      builder.AppendLine($"<td>{process.RecordsProcessed}</td>");
      builder.AppendLine("</tr>");
  }
  
  builder.AppendLine("</table></body></html>");
  
  await File.WriteAllTextAsync("etl-dashboard.html", builder.ToString());
}

public class EtlEngine
{
    private readonly IOptions<EtlSettings> _settings;
    private readonly IConnectionFactory _connectionFactory;
    private readonly ILogger<EtlEngine> _logger;
    private readonly IServiceProvider _serviceProvider;
 
    public EtlEngine(
        IOptions<EtlSettings> settings,
        IConnectionFactory connectionFactory,
        ILogger<EtlEngine> logger,
        IServiceProvider serviceProvider)
    {
        _settings = settings;
        _connectionFactory = connectionFactory;
        _logger = logger;
        _serviceProvider = serviceProvider;
    }
 
    public async Task RunAsync(string processName, CancellationToken cancellationToken = default)
    {
        var process = _settings.Value.Processes.FirstOrDefault(p => p.Name == processName)
            ?? throw new ArgumentException($"Процесс {processName} не найден в конфигурации");
 
        _logger.LogInformation("Запуск ETL-процесса {ProcessName}", processName);
        var stopwatch = Stopwatch.StartNew();
 
        try
        {
            // Извлечение данных
            var extractorType = Type.GetType(process.ExtractorType);
            var extractor = _serviceProvider.GetRequiredService(extractorType) as IDataExtractor;
            var sourceData = await extractor.ExtractAsync(process.Source, cancellationToken);
 
            // Трансформация
            var transformerType = Type.GetType(process.TransformerType);
            var transformer = _serviceProvider.GetRequiredService(transformerType) as IDataTransformer;
            var transformedData = await transformer.TransformAsync(sourceData, cancellationToken);
 
            // Загрузка
            var loaderType = Type.GetType(process.LoaderType);
            var loader = _serviceProvider.GetRequiredService(loaderType) as IDataLoader;
            await loader.LoadAsync(transformedData, process.Target, cancellationToken);
 
            stopwatch.Stop();
            _logger.LogInformation("ETL-процесс {ProcessName} завершен за {ElapsedMs}мс", 
                processName, stopwatch.ElapsedMilliseconds);
                
            // Обновление метаданных
            await UpdateEtlMetadataAsync(process, true, stopwatch.Elapsed);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Ошибка в ETL-процессе {ProcessName}", processName);
            await UpdateEtlMetadataAsync(process, false, stopwatch.Elapsed, ex.Message);
            throw;
        }
    }
 
    private async Task UpdateEtlMetadataAsync(
        ProcessSettings process, bool success, TimeSpan duration, string errorMessage = null)
    {
        using var connection = _connectionFactory.CreateMetadataConnection();
        await connection.ExecuteAsync(
            "UPDATE etl_processes SET last_run = @Now, success = @Success, " +
            "duration_seconds = @Duration, error_message = @Error",
            new 
            { 
                Now = DateTime.UtcNow, 
                Success = success, 
                Duration = duration.TotalSeconds,
                Error = errorMessage 
            });
    }
}