Работа с большими данными в C++: библиотеки и стратегии в 2026

#include <onnxruntime/core/session/onnxruntime_cxx_api.h>

class ModelInferenceService {
    Ort::Env env;
    Ort::Session session;
    Ort::MemoryInfo memory_info;
    
public:
    ModelInferenceService(const std::string& model_path) 
        : env(ORT_LOGGING_LEVEL_WARNING, "inference_service"),
          session(env, model_path.c_str(), Ort::SessionOptions{}) {
        
        // Оптимизации для конкретного железа
        Ort::SessionOptions options;
        options.SetGraphOptimizationLevel(GraphOptimizationLevel::ORT_ENABLE_ALL);
        
        #ifdef USE_CUDA
        options.AppendExecutionProvider_CUDA({});
        #elif USE_OPENVINO
        options.AppendExecutionProvider_OpenVINO({});
        #endif
        
        session = Ort::Session(env, model_path.c_str(), options);
    }
    
    // Batch inference с минимальной задержкой
    std::vector<float> predict_batch(const std::vector<std::vector<float>>& batch) {
        // Подготовка тензоров без лишних копий
        auto input_tensor = create_input_tensor(batch);
        
        // Асинхронный inference
        auto output_tensors = session.Run(
            Ort::RunOptions{nullptr},
            {"input"},
            &input_tensor, 1,
            {"output"}, 1
        );
        
        // Преобразование результатов
        return extract_results(output_tensors[0]);
    }
    
    // Streaming inference для real-time данных
    async::generator<Prediction> predict_stream(async::generator<Features> stream) {
        // Конвейерная обработка с overlapping compute/transfer
        co_await session.BeginStream();
        
        while (auto features = co_await stream.next()) {
            auto prediction = co_await session.RunAsync(features);
            co_yield prediction;
        }
        
        co_await session.EndStream();
    }
};

4.2. LibTorch 2.4+: полный цикл ML на C++

#include <torch/torch.h>
#include <torch/script.h>

// Обучение модели непосредственно на C++
class ModelTrainer {
    torch::Device device;
    
public:
    ModelTrainer() : device(torch::cuda::is_available() ? torch::kCUDA : torch::kCPU) {}
    
    torch::jit::Module train_model(const std::string& data_path) {
        // Загрузка данных через DataLoader
        auto dataset = CustomDataset(data_path)
            .map(torch::data::transforms::Normalize<>(0.5, 0.5))
            .map(torch::data::transforms::Stack<>());
        
        auto dataloader = torch::data::make_data_loader(
            std::move(dataset),
            torch::data::DataLoaderOptions().batch_size(64).workers(4)
        );
        
        // Определение модели
        auto model = NeuralNetwork();
        model->to(device);
        
        torch::optim::Adam optimizer(model->parameters(), 0.001);
        
        // Цикл обучения
        for (size_t epoch = 0; epoch < 10; ++epoch) {
            for (auto& batch : *dataloader) {
                auto [data, labels] = batch;
                data = data.to(device);
                labels = labels.to(device);
                
                optimizer.zero_grad();
                auto output = model->forward(data);
                auto loss = torch::nn::functional::cross_entropy(output, labels);
                loss.backward();
                optimizer.step();
            }
        }
        
        // Экспорт в TorchScript
        return torch::jit::trace(model, torch::randn({1, 3, 224, 224}));
    }
};

Распределенная обработка и стриминг

5.1. Ray на C++: распределенные вычисления нового поколения

#include <ray/api.h>

// Создание распределенного приложения
RAY_REMOTE(float, process_chunk, (const std::vector<float>& data));

class DistributedProcessor {
public:
    std::vector<float> process_petabyte_dataset() {
        // Автоматическое распределение по кластеру
        std::vector<ray::ObjectRef<float>> futures;
        
        for (const auto& chunk : load_data_chunks()) {
            // Асинхронный запуск задач на любом узле кластера
            futures.push_back(ray::task(process_chunk).remote(chunk));
        }
        
        // Сбор результатов с tolerance к отказам
        auto results = ray::get(futures, 
            ray::WaitOptions{
                .num_returns = futures.size(),
                .timeout_ms = 60000,
                .fetch_local = false
            });
        
        return aggregate_results(results);
    }
    
    // Stateful actors для сложных stateful вычислений
    RAY_ACTOR(AggregationActor, (std::string id), {
        float running_total = 0;
        size_t count = 0;
        
        RAY_METHOD(void, add_value, (float value), {
            running_total += value;
            count++;
        });
        
        RAY_METHOD(float, get_average, (), {
            return count > 0 ? running_total / count : 0;
        });
    });
};

5.2. Apache Kafka C++ Client с exactly-once семантикой

#include <kafka/kafka.h>

class ExactlyOnceProcessor {
    kafka::clients::KafkaProducer producer;
    kafka::clients::KafkaConsumer consumer;
    
public:
    void process_transactional() {
        // Настройка transactional producer
        producer.init_transaction();
        
        consumer.subscribe({"input-topic"});
        
        while (true) {
            auto records = consumer.poll(std::chrono::milliseconds(100));
            
            if (!records.empty()) {
                producer.begin_transaction();
                
                try {
                    std::vector<kafka::ProducerRecord> output_records;
                    
                    for (const auto& record : records) {
                        auto processed = process_record(record);
                        output_records.push_back({
                            .topic = "output-topic",
                            .value = processed
                        });
                    }
                    
                    producer.send(output_records);
                    
                    // Фиксация offset'ов и транзакции атомарно
                    producer.send_offsets_to_transaction(
                        consumer.position(consumer.assignment()),
                        consumer.group_metadata()
                    );
                    
                    producer.commit_transaction();
                    
                } catch (const std::exception& e) {
                    producer.abort_transaction();
                    throw;
                }
            }
        }
    }
};

Оптимизации для железа 2026 года

6.1. Автоматическая векторизация с помощью Highway