Обработка Big Data на C#

// Компилятор сам определит тип данных
var transactionBatch = GetMillionsOfTransactions();
// При этом тип строго определен, 
// нельзя присвоить что-то несовместимое
// transactionBatch = "строка"; // Ошибка компиляции

// До C# 14
public class Transaction
{
    public string Id { get; }
    public DateTime Date { get; }
    public decimal Amount { get; }
 
    public Transaction(string id, DateTime date, decimal amount)
    {
        Id = id;
        Date = date;
        Amount = amount;
    }
}
 
// С C# 14 и первичными конструкторами
public class Transaction(string id, DateTime date, decimal amount)
{
    public string Id { get; } = id;
    public DateTime Date { get; } = date;
    public decimal Amount { get; } = amount;
}

// Старый способ
var highValueIds = new List<string> { "txn123", "txn456", "txn789" };
 
// Новый способ с Collection Expressions
var highValueIds = ["txn123", "txn456", "txn789"];
 
// Фильтрация больших наборов данных стала нагляднее
var filtered = transactions.Where(t => highValueIds.Contains(t.Id)).ToList();

var result = from t in transactions
             where t.Amount > 1000
             group t by t.Date.Month into g
             select new { Month = g.Key, TotalAmount = g.Sum(t => t.Amount) };

var result = transactions.AsParallel()
                         .Where(t => t.Amount > 1000)
                         .GroupBy(t => t.Date.Month)
                         .Select(g => new { Month = g.Key, Total = g.Sum(t => t.Amount) });

public void ProcessChunk(Span<byte> data)
{
    // Работаем с данными без создания копий
    for (int i = 0; i < data.Length; i++)
    {
        // Обработка каждого байта
    }
}
 
// Использование
byte[] largeArray = new byte[1_000_000];
ProcessChunk(largeArray); // Никаких копий!
ProcessChunk(largeArray.AsSpan(100, 1000)); // Срез без копирования

// Иммутабельная запись транзакции
public record TransactionRecord(string Id, DateTime Date, decimal Amount);
 
// Создание новой записи на основе существующей
var original = new TransactionRecord("txn123", DateTime.Now, 100.0m);
var modified = original with { Amount = 200.0m }; // Неизменяемость!

// Без Source Generators для JSON сериализации
var options = new JsonSerializerOptions { ... };
string json = JsonSerializer.Serialize(transaction, options);
Transaction deserialized = JsonSerializer.Deserialize<Transaction>(json, options);
 
// С Source Generators
[JsonSerializable(typeof(Transaction))]
internal partial class JsonContext : JsonSerializerContext { }
 
// Более эффективная сериализация без рефлексии
string json = JsonSerializer.Serialize(transaction, JsonContext.Default.Transaction);

// Эффективная загрузка связанных данных
var query = context.Transactions
    .Where(t => t.Amount > 10000)
    .Include(t => t.Customer)
    .AsSplitQuery() // Разделение запроса для большей эффективности
    .AsNoTracking(); // Отключаем отслеживание изменений для чтения

public class ShardingDbContext : DbContext
{
    private readonly string _connectionString;
    
    public ShardingDbContext(string shardKey)
    {
        // Определяем конкретную базу данных на основе ключа шардирования
        _connectionString = DetermineConnectionString(shardKey);
    }
    
    protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
    {
        optionsBuilder.UseSqlServer(_connectionString);
    }
}

public async Task<List<AggregatedResult>> ProcessBigDataAsync(CancellationToken cancellationToken = default)
{
    // Параллельная загрузка данных из разных источников
    var task1 = _repository1.GetDataAsync(cancellationToken);
    var task2 = _repository2.GetDataAsync(cancellationToken);
    var task3 = _repository3.GetDataAsync(cancellationToken);
    
    // Ждем выполнения всех задач
    await Task.WhenAll(task1, task2, task3);
    
    // Обрабатываем результаты
    var combinedData = CombineResults(
        await task1, 
        await task2, 
        await task3);
    
    return AggregateData(combinedData);
}

public async Task ProcessLogsAsync(IEnumerable<LogFile> logFiles)
{
    await Parallel.ForEachAsync(
        logFiles, 
        new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount * 2 },
        async (file, token) =>
        {
            var content = await File.ReadAllTextAsync(file.Path, token);
            var processedData = ProcessLogContent(content);
            await SaveResultsAsync(processedData, token);
        });
}

// MongoDB и C#
public async Task<List<Transaction>> GetHighValueTransactionsAsync(decimal threshold)
{
    var filter = Builders<Transaction>.Filter.Gt(t => t.Amount, threshold);
    var sort = Builders<Transaction>.Sort.Descending(t => t.Date);
    
    return await _collection
        .Find(filter)
        .Sort(sort)
        .Limit(1000)
        .ToListAsync();
}
 
// Cassandra и C#
public async Task SaveTimeSeriesDataAsync(TimeSeriesData data)
{
    var statement = new SimpleStatement(
        "INSERT INTO time_series (id, timestamp, value) VALUES (?, ?, ?)",
        data.Id, data.Timestamp, data.Value);
    
    await _session.ExecuteAsync(statement);
}

// Параллельная обработка с использованием Redis в качестве кэша
public async Task<ProcessingResult> ProcessDataWithRedisAsync(string key)
{
    // Пробуем получить из кэша
    var cachedValue = await _redis.StringGetAsync(key);
    if (cachedValue.HasValue)
    {
        return JsonSerializer.Deserialize<ProcessingResult>(cachedValue);
    }
    
    // Выполняем дорогостоящую обработку
    var result = await ExecuteHeavyProcessingAsync(key);
    
    // Сохраняем результат в кэш
    await _redis.StringSetAsync(
        key, 
        JsonSerializer.Serialize(result),
        TimeSpan.FromHours(1));
    
    return result;
}

// Загрузка данных в Azure Data Lake
public async Task UploadToDataLakeAsync(string localFilePath, string dataLakePath)
{
    DataLakeServiceClient serviceClient = 
        new DataLakeServiceClient(_connectionString);
    
    DataLakeFileSystemClient fileSystemClient = 
        serviceClient.GetFileSystemClient("data-container");
    
    DataLakeDirectoryClient directoryClient = 
        fileSystemClient.GetDirectoryClient("processed-data");
    
    DataLakeFileClient fileClient = 
        directoryClient.GetFileClient(Path.GetFileName(dataLakePath));
    
    await using var stream = File.OpenRead(localFilePath);
    await fileClient.UploadAsync(stream, overwrite: true);
}

public async Task<DataTable> RunAnalyticsQueryAsync(string query)
{
    using var connection = new SqlConnection(_synapseConnectionString);
    await connection.OpenAsync();
    
    using var command = new SqlCommand(query, connection);
    command.CommandTimeout = 3600; // Для долгих аналитических запросов
    
    using var adapter = new SqlDataAdapter(command);
    var resultTable = new DataTable();
    adapter.Fill(resultTable);
    
    return resultTable;
}

public void ProcessCriticalDataBatch(List<Transaction> transactions)
{
    // Пытаемся заблокировать GC на время обработки
    bool noGcSucceeded = GC.TryStartNoGCRegion(
        1024 * 1024 * 100, // 100 MB
        disallowFullBlockingGC: true);
    
    try
    {
        // Критический код без прерываний на сборку мусора
        foreach (var transaction in transactions)
        {
            ProcessTransaction(transaction);
        }
    }
    finally
    {
        if (noGcSucceeded)
        {
            GC.EndNoGCRegion();
        }
    }
}

// C# пример с pattern matching и tuple
var (success, data) = await FetchDataAsync();
if (success)
{
    var result = data switch
    {
        BigDataSet { Size: > 1000000 } bds => ProcessLargeDataSet(bds),
        SmallDataSet sds => ProcessSmallDataSet(sds),
        _ => HandleUnknownDataType(data)
    };
}
 
// Аналогичный код на Java будет значительно многословнее

// Пример использования .NET for Apache Spark
var sparkSession = SparkSession.Builder().GetOrCreate();
var dataFrame = sparkSession.Read().Json("path/to/large/data.json");
 
var results = dataFrame
    .Filter("amount > 1000")
    .GroupBy("category")
    .Agg(Functions.Sum("amount").As("total_amount"))
    .OrderBy("total_amount");

// C# с явными типами
public decimal CalculateAverage(List<Transaction> transactions)
{
    return transactions.Average(t => t.Amount);
}
 
// Python с динамическими типами
def calculate_average(transactions):
    return sum(t.amount for t in transactions) / len(transactions)

// Пример использования ML.NET для прогнозирования
var mlContext = new MLContext();
var data = mlContext.Data.LoadFromTextFile<TransactionData>("data.csv", hasHeader: true);
 
var pipeline = mlContext.Transforms.CopyColumns("Label", "Amount")
    .Append(mlContext.Transforms.Categorical.OneHotEncoding("CategoryEncoded", "Category"))
    .Append(mlContext.Transforms.Concatenate("Features", "CategoryEncoded"))
    .Append(mlContext.Regression.Trainers.FastTree());
 
var model = pipeline.Fit(data);

public async Task<Stream> ReadHdfsFileAsync(string path)
{
    var client = new HttpClient();
    // Формируем URL для WebHDFS API
    var url = $"http://hdfs-namenode:50070/webhdfs/v1{path}?op=OPEN";
    
    // Делаем запрос к API
    var response = await client.GetAsync(url);
    response.EnsureSuccessStatusCode();
    
    return await response.Content.ReadAsStreamAsync();
}

public async Task UploadToHdfsAsync(string localPath, string hdfsPath)
{
    var config = new WebHdfsConfiguration
    {
        Host = "hdfs-namenode",
        Port = 50070,
        UseHttps = false
    };
    
    var client = new WebHdfsClient(config);
    
    using var stream = File.OpenRead(localPath);
    await client.CreateAsync(hdfsPath, stream, overwrite: true);
}

public class LogAnalyticsPipeline
{
    private readonly WebHdfsClient _hdfsClient;
    
    public LogAnalyticsPipeline(WebHdfsConfiguration config)
    {
        _hdfsClient = new WebHdfsClient(config);
    }
    
    public async Task ProcessDailyLogsAsync(DateTime date)
    {
        // Путь в HDFS, где хранятся логи за определенную дату
        var hdfsPath = $"/logs/{date:yyyy/MM/dd}/";
        
        // Получаем список файлов в директории
        var files = await _hdfsClient.ListStatusAsync(hdfsPath);
        
        // Параллельная обработка каждого файла
        await Parallel.ForEachAsync(files, async (file, token) =>
        {
            // Открываем файл для чтения
            using var stream = await _hdfsClient.OpenAsync(file.PathSuffix);
            using var reader = new StreamReader(stream);
            
            // Читаем логи и анализируем их
            string line;
            while ((line = await reader.ReadLineAsync()) != null)
            {
                var logEntry = ParseLogEntry(line);
                await ProcessLogEntryAsync(logEntry);
            }
        });
    }
    
    private LogEntry ParseLogEntry(string line)
    {
        // Парсинг строки лога
        // ...
        return new LogEntry();
    }
    
    private Task ProcessLogEntryAsync(LogEntry entry)
    {
        // Обработка записи лога
        // ...
        return Task.CompletedTask;
    }
}

public class LogAnalysisMapper : MapperBase
{
    public override void Map(string inputLine, MapperContext context)
    {
        var logEntry = ParseLogEntry(inputLine);
        context.EmitKeyValue(logEntry.ErrorCode.ToString(), "1");
    }
}
 
public class ErrorCountReducer : ReducerCombinerBase
{
    public override void Reduce(string key, IEnumerable<string> values, ReducerCombinerContext context)
    {
        int count = values.Sum(int.Parse);
        context.EmitKeyValue(key, count.ToString());
    }
}

// Создание SparkSession
var spark = SparkSession.Builder()
    .AppName("TransactionAnalysis")
    .Config("spark.executor.memory", "10g")
    .GetOrCreate();
 
// Загрузка данных из JSON
var transactions = spark.Read().Json("hdfs://transactions/*.json");
 
// Фильтрация и агрегация
var highValueTransactions = transactions
    .Filter("amount > 10000")
    .GroupBy("merchantId")
    .Agg(Functions.Sum("amount").As("total"), 
         Functions.Count("id").As("count"))
    .OrderBy(Functions.Desc("total"));
 
// Сохранение результатов
highValueTransactions.Write().Mode("overwrite").Parquet("hdfs://results/high-value");

// LINQ to HBase пример
var query = from row in hbaseContext.GetTable<CustomerActivity>("activities")
            where row.CustomerId == "12345" && row.Timestamp > DateTime.UtcNow.AddDays(-30)
            select new { 
                row.ActivityType, 
                row.Timestamp, 
                row.Amount 
            };
 
var recentActivities = query.ToList();

// Параллельная обработка логов с использованием PLINQ
var errorSummary = logEntries
    .AsParallel()
    .Where(log => log.Severity == LogSeverity.Error)
    .GroupBy(log => log.Component)
    .Select(group => new { 
        Component = group.Key, 
        ErrorCount = group.Count(),
        MostCommonError = group
            .GroupBy(log => log.Message)
            .OrderByDescending(g => g.Count())
            .First().Key
    })
    .ToList();

public void ProcessLargeLogFile(string filePath)
{
    using var mmf = MemoryMappedFile.CreateFromFile(
        filePath, 
        FileMode.Open, 
        null, 
        0, 
        MemoryMappedFileAccess.Read);
    
    using var accessor = mmf.CreateViewAccessor(0, 0, MemoryMappedFileAccess.Read);
    var fileSize = new FileInfo(filePath).Length;
    
    // Обрабатываем файл блоками по 1 МБ
    const int blockSize = 1024 * 1024;
    byte[] buffer = new byte[blockSize];
    
    for (long position = 0; position < fileSize; position += blockSize)
    {
        int bytesToRead = (int)Math.Min(blockSize, fileSize - position);
        accessor.ReadArray(position, buffer, 0, bytesToRead);
        
        // Обработка прочитанного блока
        ProcessDataBlock(buffer, bytesToRead);
    }
}

public static async Task ProcessTransactionsStream()
{
    // Создаем блоки для разных этапов обработки
    var readBlock = new TransformBlock<string, Transaction>(
        async filePath => await ReadTransactionFile(filePath));
    
    var validateBlock = new TransformBlock<Transaction, Transaction>(
        transaction => ValidateTransaction(transaction),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 8 });
    
    var enrichBlock = new TransformBlock<Transaction, EnrichedTransaction>(
        async transaction => await EnrichTransactionData(transaction),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });
    
    var persistBlock = new ActionBlock<EnrichedTransaction>(
        async enriched => await SaveToDatabase(enriched),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 16 });
    
    // Соединяем блоки в конвейер
    readBlock.LinkTo(validateBlock, new DataflowLinkOptions { PropagateCompletion = true });
    validateBlock.LinkTo(enrichBlock, new DataflowLinkOptions { PropagateCompletion = true });
    enrichBlock.LinkTo(persistBlock, new DataflowLinkOptions { PropagateCompletion = true });
    
    // Запускаем обработку
    foreach (var filePath in Directory.GetFiles("data", "*.json"))
    {
        await readBlock.SendAsync(filePath);
    }
    
    readBlock.Complete();
    await persistBlock.Completion;
}

// Определение интерфейса "зерна" (grain)
public interface ITransactionProcessor : IGrainWithStringKey
{
    Task<TransactionSummary> ProcessBatchAsync(List<Transaction> batch);
    Task<Dictionary<string, decimal>> GetMerchantTotalsAsync();
}
 
// Реализация зерна
public class TransactionProcessor : Grain, ITransactionProcessor
{
    private Dictionary<string, decimal> _merchantTotals = new();
    
    public Task<TransactionSummary> ProcessBatchAsync(List<Transaction> batch)
    {
        var summary = new TransactionSummary();
        
        foreach (var transaction in batch)
        {
            // Обработка транзакции и обновление агрегированных данных
            if (!_merchantTotals.ContainsKey(transaction.MerchantId))
                _merchantTotals[transaction.MerchantId] = 0;
                
            _merchantTotals[transaction.MerchantId] += transaction.Amount;
            
            summary.TotalAmount += transaction.Amount;
            summary.TransactionCount++;
        }
        
        return Task.FromResult(summary);
    }
    
    public Task<Dictionary<string, decimal>> GetMerchantTotalsAsync()
    {
        return Task.FromResult(new Dictionary<string, decimal>(_merchantTotals));
    }
}

// Определение сервиса в .proto файле
service DataProcessor {
  rpc ProcessBatch (BatchRequest) returns (BatchResponse);
}
 
// Реализация сервиса в C#
public class DataProcessorService : DataProcessor.DataProcessorBase
{
    public override async Task<BatchResponse> ProcessBatch(
        BatchRequest request, ServerCallContext context)
    {
        var results = await ProcessTransactions(request.Transactions);
        
        return new BatchResponse
        {
            ProcessedCount = results.Count,
            TotalAmount = results.Sum(r => r.Amount)
        };
    }
}

public async Task ProduceTransactionEvents(List<Transaction> transactions)
{
    var config = new ProducerConfig
    {
        BootstrapServers = "kafka1:9092,kafka2:9092",
        CompressionType = CompressionType.Snappy // Сжатие для экономии трафика
    };
    
    using var producer = new ProducerBuilder<string, string>(config)
        .SetValueSerializer(new JsonSerializer<Transaction>())
        .Build();
    
    foreach (var transaction in transactions)
    {
        // Отправляем событие в Kafka с ключом, равным ID клиента
        var result = await producer.ProduceAsync(
            "transactions-topic", 
            new Message<string, Transaction> 
            { 
                Key = transaction.CustomerId,
                Value = transaction
            });
        
        Console.WriteLine($"Delivered: {result.Offset}");
    }
}
 
public async Task ConsumeTransactionEvents()
{
    var config = new ConsumerConfig
    {
        BootstrapServers = "kafka1:9092,kafka2:9092",
        GroupId = "transaction-processors",
        AutoOffsetReset = AutoOffsetReset.Earliest
    };
    
    using var consumer = new ConsumerBuilder<string, Transaction>(config)
        .SetValueDeserializer(new JsonDeserializer<Transaction>())
        .Build();
    
    consumer.Subscribe("transactions-topic");
    
    try
    {
        while (true)
        {
            var consumeResult = consumer.Consume(TimeSpan.FromSeconds(10));
            
            if (consumeResult != null)
            {
                var transaction = consumeResult.Message.Value;
                await ProcessTransactionAsync(transaction);
                
                // Явно фиксируем обработанное сообщение
                consumer.Commit(consumeResult);
            }
        }
    }
    catch (OperationCanceledException)
    {
        // Ожидаемое исключение при закрытии потребителя
        consumer.Close();
    }
}

public async Task<ITransformer> TrainAnomalyDetectionModel(string trainingDataPath)
{
    var mlContext = new MLContext(seed: 42);
    
    // Загружаем данные из CSV
    var dataView = mlContext.Data.LoadFromTextFile<TransactionData>(
        trainingDataPath, 
        hasHeader: true, 
        separatorChar: ',');
    
    // Создаем конвейер машинного обучения
    var pipeline = mlContext.Transforms.Concatenate(
        "Features", 
        nameof(TransactionData.Amount), 
        nameof(TransactionData.Hour),
        nameof(TransactionData.DayOfWeek))
        .Append(mlContext.Transforms.NormalizeMinMax("Features"))
        .Append(mlContext.AnomalyDetection.Trainers
            .RandomizedPca(featureColumnName: "Features"));
    
    // Обучаем модель
    Console.WriteLine("Обучение модели...");
    var model = await Task.Run(() => pipeline.Fit(dataView));
    
    return model;
}
 
public IEnumerable<(Transaction Transaction, bool IsAnomaly, float Score)> DetectAnomalies(
    ITransformer model, 
    List<Transaction> transactions)
{
    var mlContext = new MLContext();
    
    // Преобразуем транзакции в формат для ML.NET
    var transactionData = transactions.Select(t => new TransactionData
    {
        TransactionId = t.Id,
        Amount = t.Amount,
        Hour = t.Timestamp.Hour,
        DayOfWeek = (int)t.Timestamp.DayOfWeek
    }).ToList();
    
    // Создаем предсказатель
    var predictionEngine = mlContext.Model.CreatePredictionEngine<TransactionData, AnomalyPrediction>(model);
    
    // Детектируем аномалии
    return transactionData.Select((data, index) =>
    {
        var prediction = predictionEngine.Predict(data);
        return (transactions[index], prediction.PredictedLabel, prediction.Score);
    });
}

public IDisposable ProcessTransactionStream()
{
    // Создаем источник данных
    var transactionSource = new Subject<Transaction>();
    
    // Настраиваем конвейер обработки
    var subscription = transactionSource
        .Buffer(TimeSpan.FromSeconds(5), 100) // Группируем по времени или количеству
        .Where(batch => batch.Any())
        .SelectMany(async batch =>
        {
            // Асинхронная обработка пакета
            return await ProcessTransactionBatchAsync(batch);
        })
        .Retry(3) // Автоматический повтор при ошибках
        .Subscribe(
            result => Console.WriteLine($"Processed {result.Count} transactions"),
            ex => Console.WriteLine($"Error: {ex.Message}"),
            () => Console.WriteLine("Stream completed")
        );
    
    // Где-то в другом месте приложения:
    // transactionSource.OnNext(new Transaction { ... });
    
    return subscription; // Возвращаем подписку для последующей отмены
}

public class TransactionMonitor
{
    private readonly SourceCache<Transaction, string> _transactions;
    private readonly ReadOnlyObservableCollection<TransactionViewModel> _transactionViews;
    
    public TransactionMonitor()
    {
        _transactions = new SourceCache<Transaction, string>(t => t.Id);
        
        // Настраиваем преобразование данных
        _transactions.Connect()
            .Transform(t => new TransactionViewModel(t))
            .Sort(SortExpressionComparer<TransactionViewModel>.Descending(t => t.Timestamp))
            .ObserveOn(RxApp.MainThreadScheduler)
            .Bind(out _transactionViews)
            .Subscribe();
    }
    
    public ReadOnlyObservableCollection<TransactionViewModel> Transactions => _transactionViews;
    
    public void AddTransaction(Transaction transaction)
    {
        _transactions.AddOrUpdate(transaction);
    }
    
    public void RemoveTransaction(string transactionId)
    {
        _transactions.Remove(transactionId);
    }
    
    public IObservable<double> GetRollingAverage(TimeSpan window)
    {
        return _transactions.Connect()
            .Filter(t => t.Timestamp >= DateTime.Now - window)
            .ToCollection()
            .Select(transactions => 
                transactions.Any() 
                    ? transactions.Average(t => t.Amount) 
                    : 0);
    }
}

public async Task SaveTransactionsToParquet(List<Transaction> transactions, string filePath)
{
    var schema = new Schema(
        new Column[] { 
            new DataColumn<string>("Id"),
            new DataColumn<DateTime>("Timestamp"),
            new DataColumn<decimal>("Amount"),
            new DataColumn<string>("MerchantId"),
            new DataColumn<string>("CustomerId")
        });
    
    using var stream = File.OpenWrite(filePath);
    using var writer = await ParquetWriter.CreateAsync(schema, stream);
    
    // Запись данных пакетами
    const int batchSize = 10000;
    for (int i = 0; i < transactions.Count; i += batchSize)
    {
        var batch = transactions.Skip(i).Take(batchSize).ToList();
        
        using var transactionGroup = writer.CreateRowGroup();
        
        await transactionGroup.WriteColumnAsync(
            new DataColumn<string>("Id", batch.Select(t => t.Id).ToArray()));
        await transactionGroup.WriteColumnAsync(
            new DataColumn<DateTime>("Timestamp", batch.Select(t => t.Timestamp).ToArray()));
        await transactionGroup.WriteColumnAsync(
            new DataColumn<decimal>("Amount", batch.Select(t => t.Amount).ToArray()));
        await transactionGroup.WriteColumnAsync(
            new DataColumn<string>("MerchantId", batch.Select(t => t.MerchantId).ToArray()));
        await transactionGroup.WriteColumnAsync(
            new DataColumn<string>("CustomerId", batch.Select(t => t.CustomerId).ToArray()));
    }
}

public class StreamProcessor
{
    private readonly IObservable<SensorReading> _sensorStream;
    private readonly ISubject<AnomalyDetection> _anomalyDetections;
    
    public StreamProcessor(IObservable<SensorReading> sensorStream)
    {
        _sensorStream = sensorStream;
        _anomalyDetections = new Subject<AnomalyDetection>();
        
        // Настраиваем конвейер обработки потока
        _sensorStream
            .Buffer(TimeSpan.FromSeconds(5), 100)
            .Select(batch => DetectAnomalies(batch))
            .Where(anomalies => anomalies.Any())
            .Subscribe(anomalies => 
            {
                foreach (var anomaly in anomalies)
                {
                    _anomalyDetections.OnNext(anomaly);
                }
            });
    }
    
    private List<AnomalyDetection> DetectAnomalies(IList<SensorReading> readings)
    {
        // Логика обнаружения аномалий
        // ...
        return new List<AnomalyDetection>();
    }
    
    public IObservable<AnomalyDetection> AnomalyDetections => _anomalyDetections;
}

public class KafkaStreamProcessor
{
    private readonly IConsumer<string, string> _consumer;
    private readonly IProducer<string, string> _producer;
    
    public KafkaStreamProcessor(string brokerList, string consumerGroup)
    {
        var consumerConfig = new ConsumerConfig
        {
            BootstrapServers = brokerList,
            GroupId = consumerGroup,
            AutoOffsetReset = AutoOffsetReset.Earliest
        };
        
        var producerConfig = new ProducerConfig
        {
            BootstrapServers = brokerList
        };
        
        _consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
        _producer = new ProducerBuilder<string, string>(producerConfig).Build();
    }
    
    public async Task StartProcessingAsync(string inputTopic, string outputTopic, CancellationToken cancellationToken)
    {
        _consumer.Subscribe(inputTopic);
        
        while (!cancellationToken.IsCancellationRequested)
        {
            var consumeResult = _consumer.Consume(cancellationToken);
            
            // Обработка сообщения
            var processedValue = ProcessMessage(consumeResult.Message.Value);
            
            // Публикация результатов
            await _producer.ProduceAsync(outputTopic, 
                new Message<string, string> 
                { 
                    Key = consumeResult.Message.Key, 
                    Value = processedValue 
                }, 
                cancellationToken);
        }
    }
    
    private string ProcessMessage(string message)
    {
        // Логика обработки сообщения
        // ...
        return $"Processed: {message}";
    }
}

[ApiController]
[Route("api/[controller]")]
public class AnalyticsController : ControllerBase
{
    private readonly IRealtimeAnalyticsService _realtimeService;
    private readonly IBatchAnalyticsService _batchService;
    
    public AnalyticsController(
        IRealtimeAnalyticsService realtimeService, 
        IBatchAnalyticsService batchService)
    {
        _realtimeService = realtimeService;
        _batchService = batchService;
    }
    
    [HttpGet("realtime/{metricId}")]
    public async Task<ActionResult<MetricValue>> GetRealtimeMetric(string metricId)
    {
        var value = await _realtimeService.GetMetricValueAsync(metricId);
        return Ok(value);
    }
    
    [HttpPost("batch")]
    public async Task<ActionResult<BatchJobResult>> StartBatchJob(BatchJobRequest request)
    {
        var jobId = await _batchService.StartJobAsync(request);
        return Accepted(new { jobId });
    }
    
    [HttpGet("batch/{jobId}")]
    public async Task<ActionResult<BatchJobStatus>> GetBatchJobStatus(string jobId)
    {
        var status = await _batchService.GetJobStatusAsync(jobId);
        return Ok(status);
    }
}

public class EventSourcingRepository<T> where T : IAggregateRoot
{
    private readonly IEventStore _eventStore;
    
    public EventSourcingRepository(IEventStore eventStore)
    {
        _eventStore = eventStore;
    }
    
    public async Task<T> GetByIdAsync(Guid id)
    {
        var events = await _eventStore.GetEventsAsync(id);
        
        if (!events.Any())
            return default;
            
        var aggregate = Activator.CreateInstance<T>();
        aggregate.LoadFromHistory(events);
        
        return aggregate;
    }
    
    public async Task SaveAsync(T aggregate)
    {
        var uncommittedEvents = aggregate.GetUncommittedEvents();
        await _eventStore.SaveEventsAsync(aggregate.Id, uncommittedEvents, aggregate.Version);
        aggregate.MarkEventsAsCommitted();
    }
}

public class CachedDataRepository : IDataRepository
{
    private readonly IDataRepository _repository;
    private readonly IDistributedCache _cache;
    private readonly ILogger<CachedDataRepository> _logger;
    
    public CachedDataRepository(
        IDataRepository repository, 
        IDistributedCache cache, 
        ILogger<CachedDataRepository> logger)
    {
        _repository = repository;
        _cache = cache;
        _logger = logger;
    }
    
    public async Task<DataResult> GetDataAsync(string queryId, QueryParameters parameters)
    {
        // Формируем ключ кэша на основе параметров запроса
        var cacheKey = $"data:{queryId}:{parameters.GetHashCode()}";
        
        // Пытаемся получить данные из кэша
        var cachedData = await _cache.GetStringAsync(cacheKey);
        
        if (cachedData != null)
        {
            _logger.LogInformation("Cache hit for {CacheKey}", cacheKey);
            return JsonSerializer.Deserialize<DataResult>(cachedData);
        }
        
        // Если данных в кэше нет, получаем их из репозитория
        _logger.LogInformation("Cache miss for {CacheKey}", cacheKey);
        var result = await _repository.GetDataAsync(queryId, parameters);
        
        // Сохраняем результат в кэш
        var cacheOptions = new DistributedCacheEntryOptions
        {
            AbsoluteExpirationRelativeToNow = TimeSpan.FromMinutes(10),
            SlidingExpiration = TimeSpan.FromMinutes(2)
        };
        
        await _cache.SetStringAsync(
            cacheKey, 
            JsonSerializer.Serialize(result), 
            cacheOptions);
        
        return result;
    }
}

public class ShardedRepository<T> where T : IEntity
{
    private readonly Dictionary<int, IRepository<T>> _shards;
    private readonly IShardResolutionStrategy _strategy;
    
    public ShardedRepository(Dictionary<int, IRepository<T>> shards, IShardResolutionStrategy strategy)
    {
        _shards = shards;
        _strategy = strategy;
    }
    
    public async Task<T> GetByIdAsync(string id)
    {
        var shardId = _strategy.GetShardId(id);
        if (!_shards.TryGetValue(shardId, out var repository))
        {
            throw new ShardNotFoundException($"Shard {shardId} not found");
        }
        
        return await repository.GetByIdAsync(id);
    }
    
    public async Task SaveAsync(T entity)
    {
        var shardId = _strategy.GetShardId(entity.Id);
        if (!_shards.TryGetValue(shardId, out var repository))
        {
            throw new ShardNotFoundException($"Shard {shardId} not found");
        }
        
        await repository.SaveAsync(entity);
    }
}

public class LambdaArchitectureService
{
    private readonly IBatchProcessor _batchProcessor;
    private readonly IStreamProcessor _streamProcessor;
    private readonly IBatchView _batchView;
    private readonly IRealtimeView _realtimeView;
    
    public LambdaArchitectureService(
        IBatchProcessor batchProcessor,
        IStreamProcessor streamProcessor,
        IBatchView batchView,
        IRealtimeView realtimeView)
    {
        _batchProcessor = batchProcessor;
        _streamProcessor = streamProcessor;
        _batchView = batchView;
        _realtimeView = realtimeView;
    }
    
    public async Task<QueryResult> QueryAsync(QueryParameters parameters)
    {
        // Получаем результаты из обоих слоев
        var batchResults = await _batchView.QueryAsync(parameters);
        var realtimeResults = await _realtimeView.QueryAsync(parameters);
        
        // Объединяем результаты
        return MergeResults(batchResults, realtimeResults);
    }
    
    public async Task ProcessDataAsync(DataBatch dataBatch)
    {
        // Отправляем данные на обработку в оба слоя
        await Task.WhenAll(
            _batchProcessor.ProcessAsync(dataBatch),
            _streamProcessor.ProcessAsync(dataBatch)
        );
    }
    
    private QueryResult MergeResults(BatchQueryResult batchResults, RealtimeQueryResult realtimeResults)
    {
        // Логика объединения результатов из обоих слоев
        // ...
        return new QueryResult();
    }
}

public class TimeSeriesProcessor
{
    private readonly ITimeSeriesDatabase _database;
    
    public TimeSeriesProcessor(ITimeSeriesDatabase database)
    {
        _database = database;
    }
    
    public async Task<AggregatedResult> AggregateTimeSeriesAsync(
        string metricName, 
        DateTime startTime, 
        DateTime endTime, 
        TimeSpan bucketSize)
    {
        // Запрашиваем сырые данные
        var rawData = await _database.QueryRawDataAsync(metricName, startTime, endTime);
        
        // Группируем по временным интервалам (бакетам)
        var buckets = rawData
            .GroupBy(point => CalculateBucket(point.Timestamp, startTime, bucketSize))
            .OrderBy(group => group.Key);
            
        // Вычисляем агрегаты для каждого интервала
        var result = new AggregatedResult();
        foreach (var bucket in buckets)
        {
            var values = bucket.Select(p => p.Value).ToList();
            result.BucketResults.Add(new BucketResult
            {
                StartTime = startTime.AddTicks(bucket.Key * bucketSize.Ticks),
                EndTime = startTime.AddTicks((bucket.Key + 1) * bucketSize.Ticks),
                Average = values.Average(),
                Min = values.Min(),
                Max = values.Max(),
                Count = values.Count
            });
        }
        
        return result;
    }
    
    private long CalculateBucket(DateTime timestamp, DateTime startTime, TimeSpan bucketSize)
    {
        var ticks = timestamp.Ticks - startTime.Ticks;
        return ticks / bucketSize.Ticks;
    }
}

public class QuantumDataProcessor
{
    public async Task<QuantumProcessingResult> ProcessWithQuantumAlgorithmAsync(
        ComplexDataSet dataSet)
    {
        // Подготовка данных для квантовой обработки
        var quantumInput = PrepareQuantumInput(dataSet);
        
        // Создание симулятора квантового компьютера
        using var qsim = new QuantumSimulator();
        
        // Запуск квантового алгоритма
        var (isOptimal, solution) = await RunGroversSearchAsync.Run(qsim, quantumInput);
        
        // Обработка результатов квантовых вычислений
        return new QuantumProcessingResult
        {
            IsOptimalSolutionFound = isOptimal,
            OptimalSolution = solution,
            ProcessingTime = DateTime.UtcNow
        };
    }
    
    private QuantumInput PrepareQuantumInput(ComplexDataSet dataSet)
    {
        // Преобразование классических данных в формат,
        // подходящий для квантовых вычислений
        // ...
        return new QuantumInput();
    }
}
 
// Квантовый алгоритм (Q#)
namespace QuantumAlgorithms {
    open Microsoft.Quantum.Canon;
    open Microsoft.Quantum.Intrinsic;
    open Microsoft.Quantum.Measurement;
    
    operation RunGroversSearch(input : QuantumInput) : (Bool, Double) {
        // Реализация алгоритма Гровера для поиска оптимального решения
        // ...
        return (true, 0.95);
    }
}

public class GraphAnalyticsService
{
    private readonly IGraphDatabase _graphDb;
    
    public GraphAnalyticsService(IGraphDatabase graphDb)
    {
        _graphDb = graphDb;
    }
    
    public async Task<List<PathResult>> FindShortestPathsAsync(
        string sourceNodeId, 
        string destinationNodeId, 
        int maxPaths = 3)
    {
        // Формируем запрос к графовой БД (например, Neo4j)
        var query = @"
            MATCH (source:Node {Id: $sourceId}),
                  (destination:Node {Id: $destId}),
                  paths = allShortestPaths((source)-[*]->(destination))
            RETURN paths
            LIMIT $maxPaths";
            
        var parameters = new Dictionary<string, object>
        {
            ["sourceId"] = sourceNodeId,
            ["destId"] = destinationNodeId,
            ["maxPaths"] = maxPaths
        };
        
        // Выполняем запрос
        var result = await _graphDb.ExecuteQueryAsync<PathResult>(query, parameters);
        return result;
    }
    
    public async Task<CommunityDetectionResult> DetectCommunitiesAsync(
        double modularity = 0.4)
    {
        // Алгоритм обнаружения сообществ (community detection)
        var query = @"
            CALL gds.louvain.stream({
                nodeProjection: 'Node',
                relationshipProjection: 'CONNECTS_TO',
                relationshipWeightProperty: 'weight',
                includeIntermediateCommunities: true,
                modularityThreshold: $modularity
            })
            YIELD nodeId, communityId, intermediateCommunityIds
            RETURN gds.util.asNode(nodeId).Id AS nodeId, 
                   communityId, 
                   intermediateCommunityIds";
                   
        var parameters = new Dictionary<string, object>
        {
            ["modularity"] = modularity
        };
        
        var results = await _graphDb.ExecuteQueryAsync<CommunityMember>(query, parameters);
        
        // Группируем узлы по сообществам
        var communities = results
            .GroupBy(r => r.CommunityId)
            .Select(g => new Community
            {
                Id = g.Key,
                MemberIds = g.Select(m => m.NodeId).ToList()
            })
            .ToList();
            
        return new CommunityDetectionResult { Communities = communities };
    }
}

public class DataLifecycleManager
{
    private readonly IStorageProvider _hotStorage; // Быстрое, дорогое хранилище
    private readonly IStorageProvider _warmStorage; // Средняя скорость/стоимость
    private readonly IStorageProvider _coldStorage; // Медленное, дешевое хранилище
    private readonly IDataAccessStatistics _statistics;
    
    public DataLifecycleManager(
        IStorageProvider hotStorage,
        IStorageProvider warmStorage,
        IStorageProvider coldStorage,
        IDataAccessStatistics statistics)
    {
        _hotStorage = hotStorage;
        _warmStorage = warmStorage;
        _coldStorage = coldStorage;
        _statistics = statistics;
    }
    
    public async Task OptimizeStorageAsync()
    {
        // Анализируем статистику доступа к данным
        var hotCandidates = await _statistics.GetFrequentlyAccessedDataAsync(TimeSpan.FromDays(7));
        var coldCandidates = await _statistics.GetRarelyAccessedDataAsync(TimeSpan.FromDays(90));
        
        // Перемещаем данные между уровнями хранения
        await Task.WhenAll(
            MoveDataToHotStorageAsync(hotCandidates),
            MoveDataToColdStorageAsync(coldCandidates)
        );
    }
    
    private async Task MoveDataToHotStorageAsync(List<DataIdentifier> dataIds)
    {
        foreach (var id in dataIds)
        {
            // Проверяем, где сейчас находятся данные
            if (await _warmStorage.ExistsAsync(id))
            {
                var data = await _warmStorage.GetDataAsync(id);
                await _hotStorage.SaveDataAsync(id, data);
                await _warmStorage.DeleteDataAsync(id);
            }
            else if (await _coldStorage.ExistsAsync(id))
            {
                var data = await _coldStorage.GetDataAsync(id);
                await _hotStorage.SaveDataAsync(id, data);
                await _coldStorage.DeleteDataAsync(id);
            }
        }
    }
    
    private async Task MoveDataToColdStorageAsync(List<DataIdentifier> dataIds)
    {
        // Аналогичная логика для перемещения данных в холодное хранилище
        // ...
    }
}