{"draftJsState":{"blocks":[{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"4gqnd"},{"data":{"embedData":{"type":"external","host":"5487fb.skillspace.ru","image":{"src":"https://avatars.dzeninfra.ru/get-zen_brief_embed/271828/pub_6983106638750c5576f90258_6983150d82cf7460a61fcb54/smart_crop_600x300","width":600,"height":300,"mainColor":"#080810"},"src":"https://5487fb.skillspace.ru/l/rust-ml-19003b","title":"ML на RUST без заморочек"}},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"atomic:embed","key":"88ig0"},{"data":{"embedData":{"type":"external","host":"gitverse.ru","image":{"src":"https://avatars.dzeninfra.ru/get-zen_brief_embed/271828/pub_6983106638750c5576f90258_69831c9195778f1892af5c0c/smart_crop_600x300","width":180,"height":180,"mainColor":"#F8F8F8"},"src":"https://gitverse.ru/nicktretyakov1/CDCML","title":"nicktretyakov1/CDCML | Gitverse"}},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"atomic:embed","key":"a8bhu"},{"data":{"embedData":{"type":"external","host":"vk.com","image":{"src":"https://avatars.dzeninfra.ru/get-zen_brief_embed/271828/pub_6983106638750c5576f90258_69c150b556dd356c82ba87f8/smart_crop_600x300","width":240,"height":240,"mainColor":"#080808"},"src":"https://vk.com/rustmlclub","title":"Rust ML без заморочек"}},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"atomic:embed","key":"7obmq"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"c1qot"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Для чего нужна данная статья?:","type":"header-three","key":"8tb5n"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":26,"offset":17,"style":"BOLD"},{"length":1,"offset":75,"style":"BOLD"}],"text":"Написать код CDC real-time data processing обладающий следующими свойствами:","type":"unstyled","key":"fv612"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":16,"offset":0,"style":"BOLD"}],"text":"Масштабируемость: тысячи партиций (добавляйте readers), авто-backpressure предотвращает OOM.","type":"unordered-list-item","key":"dt5q5"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":15,"offset":0,"style":"BOLD"}],"text":"Тысячи партиций: данные делятся на множество независимых частей (партиций), каждая из которых обрабатывается отдельно. Например, как в супермаркете: вместо одной кассы — десятки, и каждый покупатель идёт к своей.","type":"unstyled","key":"6gjmq"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"Авто-backpressure: если система не успевает обрабатывать данные, она автоматически замедляет их поступление, чтобы не перегрузиться (как если бы кассир просил покупателей подождать, пока освободится место на ленте).","type":"unstyled","key":"8pmfh"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"2r3l"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":9,"offset":0,"style":"BOLD"}],"text":"Интеллект: ML-батчинг на GPU детектирует аномалии в реал-тайм, роутинг в quarantine.","type":"unordered-list-item","key":"b9fav"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"ML-батчинг на GPU: данные собираются в небольшие пакеты (батчи) и анализируются на видеокартах (GPU), что ускоряет процесс. Например, как если бы вы сортировали фрукты не по одному, а сразу горстями.","type":"unstyled","key":"fgj1e"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":23,"offset":0,"style":"BOLD"}],"text":"Детектирование аномалий: система находит необычные события (например, внезапный скачок температуры на производстве или подозрительная транзакция в банке).","type":"unstyled","key":"ajndg"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":20,"offset":0,"style":"BOLD"}],"text":"Роутинг в quarantine: подозрительные данные отправляются на дополнительную проверку (как в аэропорту: подозрительный багаж проверяют отдельно).","type":"unstyled","key":"cd371"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"fbrpt"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":10,"offset":0,"style":"BOLD"}],"text":"Надёжность: exactly-once (Kafka transactions), checkpointing (resume после рестарта), error handling с метриками.","type":"unordered-list-item","key":"6jjjl"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":33,"offset":0,"style":"BOLD"}],"text":"Exactly-once (Kafka transactions): каждое событие обрабатывается ровно один раз, без дубликатов или потерь. Например, как если бы вы платили за товар один раз, а не дважды или вообще не платили.","type":"unstyled","key":"79tpg"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":13,"offset":0,"style":"BOLD"}],"text":"Checkpointing: система сохраняет состояние обработки, чтобы после перезапуска продолжить с того же места (как закладка в книге).","type":"unstyled","key":"9ch7k"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":26,"offset":0,"style":"BOLD"}],"text":"Error handling с метриками: ошибки фиксируются и анализируются, чтобы быстро находить и исправлять проблемы.","type":"unstyled","key":"ec7ia"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"6l1ie"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":16,"offset":0,"style":"BOLD"}],"text":"Поддерживаемость: SOLID + Hexagonal — новые источники/синки/ML-модели добавляются как плагины без изменения core/orchestrator.","type":"unordered-list-item","key":"5m3gt"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":29,"offset":0,"style":"BOLD"}],"text":"SOLID + Hexagonal архитектура: код разбит на независимые модули, которые можно менять или добавлять, не ломая всю систему. Например, как конструктор Lego: можно заменить одну деталь, не разбирая всю постройку.","type":"unstyled","key":"1tctr"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Плагины: новые источники данных, модели или способы сохранения добавляются как отдельные модули (как установка нового приложения на телефон).","type":"unstyled","key":"p8fc"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"4b71g"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":11,"offset":0,"style":"BOLD"}],"text":"Performance: parallel batch inference (256+ событий за инференс), staged parallelism (CPU/GPU/I/O отдельно).","type":"unordered-list-item","key":"83llr"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":24,"offset":0,"style":"BOLD"}],"text":"Parallel batch inference: данные анализируются параллельно большими партиями (например, 256 событий за один раз), что ускоряет процесс.","type":"unstyled","key":"f6coq"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":18,"offset":0,"style":"BOLD"}],"text":"Staged parallelism: разные этапы обработки (CPU, GPU, ввод/вывод) работают параллельно, не мешая друг другу. Например, как на кухне: пока один повар режет овощи, другой уже жарит мясо, а третий моет посуду.","type":"unstyled","key":"6dmp2"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"6igjt"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"96oae"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":19,"offset":11,"style":"BOLD"}],"text":"Архитектура (Ports & Adapters):","type":"header-two","key":"d5v9e"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":4,"offset":0,"style":"BOLD"}],"text":"Core: чистая доменная логика (типы изменений, ошибки).","type":"unordered-list-item","key":"m7mm"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"},{"length":7,"offset":98,"style":"BOLD"},{"length":3,"offset":53,"style":"ITALIC"},{"length":3,"offset":82,"style":"ITALIC"}],"text":"Простыми словами: Это сердце программы, где описано, что она должна делать, но не как это делать.\nПример:\nПредставьте, что вы пишете программу для банка. В Core будет логика: \"Если на счёте достаточно денег, разрешить перевод\". Но не будет кода, который реально списывает деньги с базы данных.","type":"unstyled","key":"fojuk"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":5,"offset":0,"style":"BOLD"}],"text":"Ports: абстрактные трейты (интерфейсы) — обеспечивают DIP (Dependency Inversion) и ISP (Interface Segregation).","type":"unordered-list-item","key":"2t92j"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"},{"length":7,"offset":105,"style":"BOLD"}],"text":"Простыми словами: Это \"контракты\" или правила, по которым другие части программы должны общаться с Core.\nПример:\nПродолжим банковский пример. Port может быть интерфейсом \"ПереводДенег\" с методом выполнитьПеревод(сумма). Конкретная реализация (например, через базу данных или API) будет подключаться к этому интерфейсу.","type":"unstyled","key":"fpl7r"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":8,"offset":0,"style":"BOLD"}],"text":"Adapters: конкретные реализации (PostgreSQL polling reader, simple transformer, Torch-based ML enricher с GPU-батчингом, Kafka transactional writer, sled checkpoint).","type":"unordered-list-item","key":"2eup9"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"},{"length":7,"offset":88,"style":"BOLD"}],"text":"Простыми словами: Это реальные \"детали\", которые выполняют работу по правилам из Ports.\nПример:\nАдаптер для базы данных PostgreSQL будет реализовывать метод выполнитьПеревод(сумма) так, чтобы деньги реально списывались из базы.","type":"unstyled","key":"avh8o"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":12,"offset":0,"style":"BOLD"}],"text":"Orchestrator: координация всего пайплайна — многопоточный, асинхронный, с backpressure и интеллектуальным роутингом аномалий.","type":"unordered-list-item","key":"cvt61"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"},{"length":7,"offset":114,"style":"BOLD"}],"text":"Простыми словами: Это \"дирижёр\", который управляет, в каком порядке и как быстро выполняются все части программы.\nПример:\nВ банке Orchestrator может следить, чтобы переводы не накапливались в очереди, а обрабатывались быстро и без ошибок, даже если одновременно пришло много запросов.","type":"unstyled","key":"a1c9e"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Metrics: детальная observability для production (latency, throughput, queues, errors).","type":"unordered-list-item","key":"pcd2"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"},{"length":7,"offset":127,"style":"BOLD"}],"text":"Простыми словами: Это инструменты, которые показывают, как работает программа в реальном времени (скорость, ошибки, загрузка).\nПример:\nКак в банке на экране оператора видно, сколько переводов в секунду проходит, нет ли задержек или ошибок.","type":"unstyled","key":"bp76m"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Проект использует Tokio для асинхронности, tch для ML-инференса на GPU/CPU, rdkafka для exactly-once, sled для checkpointing.","type":"unstyled","key":"83ris"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":5,"offset":0,"style":"BOLD"}],"text":"Tokio — библиотека для асинхронной работы (чтобы программа могла делать много дел одновременно).","type":"unordered-list-item","key":"fisnd"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":3,"offset":0,"style":"BOLD"}],"text":"tch — обёртка для PyTorch, чтобы запускать машинное обучение на GPU/CPU.","type":"unordered-list-item","key":"f7es8"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"rdkafka — библиотека для работы с Kafka (система для передачи сообщений между программами).","type":"unordered-list-item","key":"8bn1r"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":4,"offset":0,"style":"BOLD"}],"text":"sled — база данных для сохранения состояния (например, чтобы запомнить, какие данные уже обработали).","type":"unordered-list-item","key":"577tj"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"99v15"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"1. Core модуль (src/core/)","type":"header-three","key":"2a9vj"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":18,"offset":0,"style":"BOLD"}],"text":"src/core/change.rs — доменные типы (SRP: только данные изменений).","type":"unstyled","key":"fd3r8"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"RawChange: сырое событие из источника (id, partition_key для шардинга, operation, payload как JSON, timestamp). Это исходное, необработанное событие, которое приходит из какого-то источника данных (например, базы данных, лога, очереди сообщений). Оно содержит минимально необходимую информацию для дальнейшей обработки.","type":"unordered-list-item","key":"2ismo"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Operation: enum для Insert/Update/Delete. Это событие, которое уже преобразовано в удобный для бизнеса формат. Оно содержит не сырые данные, а структурированную информацию, понятную для дальнейшей логики (например, для машинного обучения).","type":"unordered-list-item","key":"du46b"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"ProcessedChange: enum (расширяемый) — сейчас только Order(OrderChange) с бизнес-данными и synthetic_features для ML. Это событие, которое уже преобразовано в удобный для бизнеса формат. Оно содержит не сырые данные, а структурированную информацию, понятную для дальнейшей логики (например, для машинного обучения).","type":"unordered-list-item","key":"ckor8"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":11,"offset":0,"style":"BOLD"}],"text":"Что внутри?","type":"unstyled","key":"9aq28"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Сейчас упоминается только Order(OrderChange) — то есть изменения, связанные с заказами, с бизнес-данными (например, сумма заказа, статус, список товаров).","type":"unstyled","key":"e7qtm"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"synthetic_features — искусственно созданные признаки для машинного обучения (например, \"средняя сумма заказов пользователя за месяц\").","type":"unstyled","key":"13ura"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"aa0ap"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"EnrichedChange: обогащённое изменение (inner + anomaly_score + classification от ML). Это событие, которое уже прошло через машинное обучение или другие системы анализа. Оно содержит не только бизнес-данные, но и результаты анализа (например, оценку аномальности, классификацию).","type":"unordered-list-item","key":"sto"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":11,"offset":0,"style":"BOLD"}],"text":"Что внутри?","type":"unstyled","key":"6ma48"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"inner — исходное ProcessedChange.","type":"unordered-list-item","key":"6ns0q"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"anomaly_score — оценка, насколько это событие необычное (например, 0.9 — очень подозрительно).","type":"unordered-list-item","key":"a8qqc"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"classification — результат классификации (например, \"мошенничество\", \"нормальный заказ\")","type":"unordered-list-item","key":"ec6ck"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"4nkht"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Это обеспечивает OCP (Open/Closed): новые типы изменений добавляются без изменения существующего кода. Это принцип проектирования: код должен быть открыт для расширения, но закрыт для изменения. То есть, чтобы добавить новый тип изменения (например, не только заказы, но и возвраты), не нужно менять уже существующий код — достаточно добавить новый вариант в enum.","type":"unstyled","key":"bgm2u"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nЕсли завтра понадобится обрабатывать не только заказы, но и возвраты, можно просто добавить новый вариант в enum:","type":"unstyled","key":"fa9je"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":12,"offset":81,"style":"ITALIC"}],"text":"rustCopyenum ProcessedChange {\n Order(OrderChange),\n Return(ReturnChange), // новый тип\n}","type":"unstyled","key":"dg049"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"При этом старый код для обработки заказов не меняется.","type":"unstyled","key":"fuks7"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"3eaib"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"src/core/error.rs : thiserror-derive enum с From для всех внешних ошибок (IO, Postgres, Sled, Tch, Kafka) — цепочка ошибок для надёжности. Это файл, где описаны все возможные ошибки, которые могут возникнуть в системе (например, ошибки чтения/записи, базы данных, очередей). Используется библиотека thiserror для удобного создания цепочек ошибок (чтобы понимать, какая ошибка к какой привела).","type":"unstyled","key":"e530l"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nЕсли не удалось записать данные в базу, ошибка может выглядеть так:","type":"unstyled","key":"5aoog"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"CopyError: DatabaseError { source: IOError { ... } }","type":"unstyled","key":"1r577"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Это помогает быстро найти причину проблемы.","type":"unstyled","key":"82m1j"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Итоговая схема:","type":"header-three","key":"c4l10"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"CopyRawChange (сырые данные)\n → ProcessedChange (бизнес-логика)\n → EnrichedChange (ML-анализ)","type":"unstyled","key":"587cl"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Каждый этап добавляет новые данные, не меняя предыдущие.","type":"unstyled","key":"fkgnk"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"b5d93"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"2. Ports (src/ports/) — SOLID в чистом виде","type":"header-three","key":"do5sp"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Трейты — аналоги интерфейсов Java/C#.","type":"unstyled","key":"q7gb"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":9,"offset":0,"style":"BOLD"},{"length":17,"offset":108,"style":"BOLD"}],"text":"reader.rs: BatchChangeReader (наследует ChangeReader) — чтение батчами для эффективности (batch inference). Простыми словами:\nЭто компонент, который читает данные не по одному, а большими порциями (батчами), чтобы ускорить обработку. Представьте, что вы загружаете фото в Instagram: вместо того, чтобы отправлять их по одному, вы отправляете сразу 10 — так быстрее и эффективнее.","type":"unordered-list-item","key":"1arnr"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nЕсли у вас есть 1000 записей в базе данных, вместо того, чтобы читать каждую по отдельности, вы читаете сразу по 100 записей за раз.","type":"unstyled","key":"1sith"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"8lmga"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":12,"offset":0,"style":"BOLD"},{"length":17,"offset":81,"style":"BOLD"}],"text":"processor.rs: ChangeProcessor — трансформация raw → processed (валидация, фичи). Простыми словами:\nЭто компонент, который берёт \"сырые\" данные (например, текст из формы обратной связи) и преобразует их в удобный формат: проверяет на ошибки, добавляет нужные поля, очищает от мусора.","type":"unordered-list-item","key":"91bbn"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nВы заполняете анкету на сайте: ваш возраст — \"двадцать пять\". Процессор преобразует это в число 25, проверяет, что это корректный возраст, и добавляет метку \"взрослый\".","type":"unstyled","key":"99ad"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"1088f"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":11,"offset":0,"style":"BOLD"},{"length":17,"offset":75,"style":"BOLD"}],"text":"enricher.rs: ChangeEnricher — батчевое обогащение ML (parallel inference). Простыми словами:\nЭто компонент, который добавляет к данным новую информацию с помощью машинного обучения. Например, по фотографии определяет, что на ней кот, и добавляет тег \"кот\". Работает параллельно — сразу с несколькими данными, чтобы быстрее.","type":"unordered-list-item","key":"16ohg"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nУ вас есть список товаров. Обогатитель анализирует описания и добавляет теги: \"электроника\", \"бытовая техника\", \"акция\".","type":"unstyled","key":"29kn3"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"3mlnk"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":9,"offset":0,"style":"BOLD"},{"length":17,"offset":69,"style":"BOLD"}],"text":"writer.rs: ChangeWriter — батчевая запись + commit для exactly-once. Простыми словами:\nЭто компонент, который сохраняет обработанные данные большими порциями и гарантирует, что каждая запись будет сохранена ровно один раз — без дубликатов и потерь.","type":"unordered-list-item","key":"11ru9"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nВы переводите деньги с карты на карту. Важно, чтобы сумма списалась и зачислилась ровно один раз, а не дважды или ни разу.","type":"unstyled","key":"2jh73"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"3g1m2"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":102,"style":"BOLD"}],"text":"Все трейты async, Send + Sync — для многопоточности. LSP соблюдается: все реализации взаимозаменяемы. Простыми словами:","type":"unstyled","key":"1n7gq"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":5,"offset":0,"style":"BOLD"}],"text":"async — компоненты работают асинхронно, то есть не блокируют друг друга, пока ждут ответа (например, пока один компонент ждёт данные из базы, другой может заниматься обработкой).","type":"unordered-list-item","key":"25j1c"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":11,"offset":0,"style":"BOLD"}],"text":"Send + Sync — компоненты можно безопасно использовать в нескольких потоках одновременно, как несколько поваров на кухне, которые не мешают друг другу.","type":"unordered-list-item","key":"8of2e"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nВы загружаете видео на YouTube и одновременно редактируете описание — оба процесса идут параллельно.","type":"unstyled","key":"b4u78"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"2o76k"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"3. Adapters — конкретные плагины (OCP: расширяемость)","type":"header-three","key":"7r566"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":19,"offset":0,"style":"BOLD"}],"text":"postgres_polling.rs (partition-level reader):","type":"unstyled","key":"36mec"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Подключается к Postgres, поллит по id > last_id (incremental).","type":"unordered-list-item","key":"ff873"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"last_id в Mutex для thread-safety.","type":"unordered-list-item","key":"8q3ct"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"read_batch возвращает Vec — батч для эффективности.","type":"unordered-list-item","key":"cgkin"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"poll_interval для idle.","type":"unordered-list-item","key":"cbrfl"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":11,"offset":0,"style":"BOLD"}],"text":"Что делает:\nМодуль постоянно проверяет базу данных PostgreSQL на наличие новых записей, начиная с последнего сохранённого id.","type":"unstyled","key":"7tdaa"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"Ключевые понятия:","type":"unstyled","key":"a9pev"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":44,"offset":0,"style":"BOLD"},{"length":7,"offset":222,"style":"ITALIC"}],"text":"Incremental polling (инкрементальный опрос):\nВместо того, чтобы каждый раз скачивать все данные, код запоминает последний обработанный id (например, last_id = 1000) и в следующий раз запрашивает только записи с id > 1000.\nПример: Если в базе появились записи с id 1001, 1002, 1003 — код обработает только их.","type":"unordered-list-item","key":"4jjmq"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":16,"offset":0,"style":"BOLD"},{"length":7,"offset":194,"style":"ITALIC"}],"text":"Mutex (мьютекс):\nЭто \"замок\" для защиты переменной last_id от одновременного изменения несколькими потоками. Без мьютекса два потока могут одновременно изменить last_id, что приведёт к ошибкам.\nПример: Представьте, что два человека пытаются одновременно записать номер последней обработанной записи в блокнот — без замка они перезапишут друг друга.","type":"unordered-list-item","key":"9dt4i"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":28,"offset":0,"style":"BOLD"},{"length":7,"offset":155,"style":"ITALIC"}],"text":"read_batch (чтение пачками):\nВместо обработки каждой записи по отдельности, код читает сразу несколько записей (например, 100 штук) — это ускоряет работу.\nПример: Как если бы вы несли книги из библиотеки не по одной, а сразу стопкой.","type":"unordered-list-item","key":"e1en6"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":32,"offset":0,"style":"BOLD"}],"text":"poll_interval (интервал опроса):\nЕсли новых данных нет, код не будет постоянно нагружать базу — он делает паузу (например, 1 секунду) перед следующим запросом.","type":"unordered-list-item","key":"7tmfh"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"8mn4r"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":21,"offset":0,"style":"BOLD"}],"text":"simple_transformer.rs:","type":"unstyled","key":"ev9a7"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Process: парсит payload, валидация (negative amount → error), генерирует synthetic_features (log10(amount), normalized user_id, timestamp) для ML.","type":"unordered-list-item","key":"6gmj2"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Возвращает ProcessedChange::Order.","type":"unordered-list-item","key":"dohpv"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":11,"offset":0,"style":"BOLD"}],"text":"Что делает:\nМодуль парсит и валидирует данные, а также готовит их для машинного обучения.","type":"unstyled","key":"fvnp7"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"Ключевые понятия:","type":"unstyled","key":"5rjqo"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":32,"offset":0,"style":"BOLD"},{"length":7,"offset":125,"style":"ITALIC"}],"text":"Парсинг payload (разбор данных):\nКод извлекает из сырых данных нужные поля (например, сумму заказа, id пользователя, время).\nПример: Из строки {\"order_id\": 123, \"amount\": 500} извлекаются order_id и amount.","type":"unordered-list-item","key":"2ler7"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":28,"offset":0,"style":"BOLD"},{"length":7,"offset":86,"style":"ITALIC"}],"text":"Валидация (проверка данных):\nЕсли сумма заказа отрицательная — код возвращает ошибку.\nПример: Если amount = -100, это ошибка — сумма не может быть отрицательной.","type":"unordered-list-item","key":"bck4j"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":44,"offset":0,"style":"BOLD"},{"length":8,"offset":132,"style":"ITALIC"}],"text":"Synthetic features (синтетические признаки):\nКод создаёт новые признаки на основе исходных данных, чтобы улучшить работу ML-модели.\nПримеры:log10(amount) — логарифм суммы (чтобы большие суммы не искажали модель).\nnormalized user_id — приведение id пользователя к диапазону 0–1.\ntimestamp — время заказа в удобном формате.\n","type":"unordered-list-item","key":"35nm3"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"clfac"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":14,"offset":0,"style":"BOLD"}],"text":"ml_enricher.rs (многопоточный ML с батчингом):","type":"unstyled","key":"6uacj"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"TorchModelProvider: загружает TorchScript модель один раз, set_eval, device (GPU if available).","type":"unordered-list-item","key":"fdv4g"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"extract_features: берёт synthetic_features из OrderChange.","type":"unordered-list-item","key":"923uc"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"enrich_batch:Собирает фичи в Vec>.\nStack в Tensor на device (GPU для скорости).\nforward_ts — инференс всего батча параллельно на GPU.\nto_cpu + f_values — результаты.\nПарсит anomaly_score и classification.\n","type":"unordered-list-item","key":"2aq6j"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Максимальная сложность: dynamic batch size, GPU acceleration, fallback.","type":"unordered-list-item","key":"e26if"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":11,"offset":0,"style":"BOLD"}],"text":"Что делает:\nМодуль загружает модель машинного обучения и применяет её к данным, используя GPU для ускорения.","type":"unstyled","key":"ekugb"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"Ключевые понятия:","type":"unstyled","key":"12i10"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":19,"offset":0,"style":"BOLD"},{"length":7,"offset":122,"style":"ITALIC"}],"text":"TorchScript модель:\nМодель машинного обучения, сохранённая в формате TorchScript (можно быстро загрузить и использовать).\nПример: Как если бы вы сохранили рецепт торта в книге, чтобы потом быстро его испечь.","type":"unordered-list-item","key":"7o2vg"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":30,"offset":0,"style":"BOLD"},{"length":7,"offset":165,"style":"ITALIC"}],"text":"Batching (пакетная обработка):\nВместо того, чтобы обрабатывать каждую запись по отдельности, код собирает их в пакеты (например, по 32 записи) и обрабатывает сразу.\nПример: Как если бы вы пекли не один пирог, а сразу несколько на одном противне.","type":"unordered-list-item","key":"38qek"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":36,"offset":0,"style":"BOLD"},{"length":7,"offset":120,"style":"ITALIC"}],"text":"GPU acceleration (ускорение на GPU):\nМодель работает на видеокарте (GPU), что намного быстрее, чем на процессоре (CPU).\nПример: Как если бы вы считали на калькуляторе вместо счётов на бумаге.","type":"unordered-list-item","key":"dopjg"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":48,"offset":0,"style":"BOLD"},{"length":7,"offset":113,"style":"ITALIC"}],"text":"Dynamic batch size (динамический размер пакета):\nРазмер пакета может меняться в зависимости от доступной памяти.\nПример: Если памяти мало — пакет из 10 записей, если много — из 100.","type":"unordered-list-item","key":"1iugs"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"},{"length":7,"offset":69,"style":"ITALIC"}],"text":"Fallback (откат):\nЕсли GPU недоступен, модель может работать на CPU.\nПример: Если электричество отключили, вы можете готовить на газовой плите.","type":"unordered-list-item","key":"5tkd5"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"420n8"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":22,"offset":0,"style":"BOLD"}],"text":"kafka_transactional.rs (exactly-once writer):","type":"unstyled","key":"d4202"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"FutureProducer с transactional.id и idempotence.","type":"unordered-list-item","key":"12bnq"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"init_transactions при создании.","type":"unordered-list-item","key":"bmiic"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"write_batch: begin_transaction → send с key (order_id) → commit_transaction.","type":"unordered-list-item","key":"fgnos"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Обрабатывает ошибки через CdcError::Kafka.","type":"unordered-list-item","key":"6ght2"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":11,"offset":0,"style":"BOLD"}],"text":"Что делает:\nМодуль записывает данные в Kafka так, чтобы каждая запись была доставлена ровно один раз (exactly-once).","type":"unstyled","key":"5vlfi"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"Ключевые понятия:","type":"unstyled","key":"ci2li"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":49,"offset":0,"style":"BOLD"},{"length":7,"offset":151,"style":"ITALIC"}],"text":"Transactional producer (транзакционный продюсер):\nКод начинает транзакцию, отправляет данные, и только если всё прошло успешно — фиксирует транзакцию.\nПример: Как если бы вы переводили деньги: сначала проверяете баланс, потом списываете, и только потом подтверждаете перевод.","type":"unordered-list-item","key":"3j62h"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":30,"offset":0,"style":"BOLD"},{"length":7,"offset":102,"style":"ITALIC"}],"text":"Idempotence (идempотентность):\nЕсли одна и та же запись отправится дважды, Kafka не создаст дубликат.\nПример: Если вы дважды нажали \"Отправить\" в мессенджере, сообщение отправится только один раз.","type":"unordered-list-item","key":"ehbu5"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":15,"offset":0,"style":"BOLD"}],"text":"Key (order_id):\nКаждая запись имеет уникальный ключ (например, order_id), чтобы Kafka могла правильно распределять данные по партициям.","type":"unordered-list-item","key":"b2q5a"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"fps8o"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":13,"offset":0,"style":"BOLD"}],"text":"checkpoint.rs:","type":"unstyled","key":"9jf3n"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Sled Db (embedded KV) для offsets по partition_key.","type":"unordered-list-item","key":"f3g58"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"save/load_offset с to_be_bytes — атомарно и надёжно.","type":"unordered-list-item","key":"8scdk"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":11,"offset":0,"style":"BOLD"}],"text":"Что делает:\nМодуль сохраняет, до какого id или offset данные уже обработаны, чтобы при перезапуске не начинать с нуля.","type":"unstyled","key":"3gk0f"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"Ключевые понятия:","type":"unstyled","key":"6c8qr"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":33,"offset":0,"style":"BOLD"},{"length":7,"offset":120,"style":"ITALIC"}],"text":"Sled Db (встроенная база данных):\nЭто лёгкая база данных, которая хранит данные прямо на диске, без отдельного сервера.\nПример: Как если бы вы записывали прогресс чтения книги на листочке, который всегда лежит рядом.","type":"unordered-list-item","key":"5efr0"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":18,"offset":0,"style":"BOLD"},{"length":7,"offset":86,"style":"ITALIC"}],"text":"Offset (смещение):\nЭто номер последней обработанной записи (например, offset = 1000).\nПример: Если вы читаете книгу и запомнили, что дочитали до 100 страницы — это ваш offset.","type":"unordered-list-item","key":"9gbls"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":32,"offset":0,"style":"BOLD"},{"length":7,"offset":115,"style":"ITALIC"}],"text":"Атомарность (atomic operations):\nСохранение offset происходит целиком и сразу, без риска потерять данные при сбое.\nПример: Как если бы вы записывали номер страницы одним движением ручки, а не по цифрам.","type":"unordered-list-item","key":"b29sp"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"ap2pi"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"4. Orchestrator — многопоточный staged pipeline","type":"header-three","key":"ttlr"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":5,"offset":0,"style":"BOLD"}],"text":"run():","type":"unstyled","key":"8ukjv"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Создаёт bounded mpsc channels (backpressure: max_inflight_batches ограничивает память).","type":"unordered-list-item","key":"50egq"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"Простыми словами:\nЭто как очередь в магазине, где количество мест ограничено. Если очередь заполнилась, новые покупатели не могут встать, пока кто-то не освободит место. Такой подход предотвращает перегрузку системы.","type":"unstyled","key":"9s02"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nПредставьте, что у вас есть корзина для почты. Вы можете положить туда только 10 писем. Если корзина заполнена, новые письма не будут приниматься, пока вы не обработаете хотя бы одно.","type":"unstyled","key":"704dr"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"ae166"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":12,"offset":0,"style":"BOLD"}],"text":"Reader stage: по задаче на reader (partition parallelism) — бесконечный loop, read_batch, метрики (inc_changes_read), send в tx_raw.","type":"unordered-list-item","key":"3mbst"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"Простыми словами:\nЭто часть программы, которая читает данные из какого-то источника (например, из базы данных или файла) и отправляет их дальше на обработку.","type":"unstyled","key":"5hcfu"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nПредставьте, что у вас есть книга, и вы читаете её по одной странице, а потом передаёте прочитанное другу для анализа.","type":"unstyled","key":"101rc"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"abkr1"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":16,"offset":0,"style":"BOLD"}],"text":"Processing stage: parallelism_per_stage задач — recv raw_batch → parallel transform (join_all) → enrich_batch с latency метрикой → partition на normal/quarantine по threshold → send в tx_enriched.","type":"unordered-list-item","key":"1src1"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"Простыми словами:\nЗдесь данные, полученные от читателей, обрабатываются: трансформируются, обогащаются, проверяются на ошибки.","type":"unstyled","key":"2mfaf"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nВаш друг получает страницы книги, выделяет главные мысли, добавляет свои комментарии и сортирует страницы по темам.","type":"unstyled","key":"ati6i"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"sf76"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":12,"offset":0,"style":"BOLD"}],"text":"Writer stage: отдельные задачи — recv, detect anomaly in batch, choose writer (normal/quarantine), write_batch.","type":"unordered-list-item","key":"4ttvl"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"Простыми словами:\nЭто финальный этап, где обработанные данные сохраняются в нужное место (например, в другую базу данных или файл).","type":"unstyled","key":"ejjhq"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nПосле анализа страниц ваш друг записывает итоговые выводы в тетрадь.","type":"unstyled","key":"1f9op"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"f375d"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Graceful shutdown по Ctrl+C или завершению задач.","type":"unordered-list-item","key":"dh2ia"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Метрики на всех этапах: queues size, latency, counts, errors.","type":"unordered-list-item","key":"84uui"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":39,"offset":4,"style":"BOLD"}],"text":"Это pipeline parallelism + data parallelism: readers I/O-bound, processing CPU/GPU-bound, writers I/O-bound.","type":"unstyled","key":"77bjl"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":17,"offset":0,"style":"BOLD"}],"text":"Простыми словами:\nЭто как конвейер на заводе: каждый работник выполняет свою часть работы, и все части работают одновременно, но независимо друг от друга.","type":"unstyled","key":"7fmhl"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nНа заводе по сборке машин один рабочий ставит колёса, другой — двигатель, третий — красит кузов. Все работают одновременно, но каждый над своей задачей.","type":"unstyled","key":"632o3"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"44g2o"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"5. Metrics — полная observability","type":"header-three","key":"a34lm"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Prometheus exporter: describe всех ключевых метрик (counters для counts, gauges для dynamic: queues, latency).\nФункции inc_* и set_* — вызываются в orchestrator для реал-тайм мониторинга (autoscaling в K8s по latency/queues).","type":"unstyled","key":"plv5"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":8,"offset":124,"style":"BOLD"}],"text":"Prometheus — это система для сбора и хранения метрик (чисел, которые показывают, как работает ваша программа или сервер). А exporter — это специальная программа, которая собирает эти метрики из вашего приложения и отдаёт их Prometheus.","type":"unstyled","key":"f20a9"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Пример:\nДопустим, у вас есть сайт. Вы хотите знать, сколько людей заходит на сайт каждую секунду. Exporter будет считать этих людей и отправлять данные в Prometheus.","type":"unstyled","key":"62mpj"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":8,"offset":0,"style":"BOLD"},{"length":7,"offset":92,"style":"BOLD"}],"text":"Counters — счётчики, которые только увеличиваются (например, сколько раз что-то произошло).\nПример: Количество заказов на сайте. Каждый новый заказ увеличивает счётчик на 1.","type":"unordered-list-item","key":"1daud"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":6,"offset":0,"style":"BOLD"},{"length":7,"offset":130,"style":"BOLD"}],"text":"Gauges — метрики, которые могут как увеличиваться, так и уменьшаться (например, текущая температура, количество людей в очереди).\nПример: Количество людей, одновременно находящихся на сайте. Оно может то расти, то падать.","type":"unordered-list-item","key":"247eq"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":6,"offset":0,"style":"BOLD"},{"length":7,"offset":53,"style":"BOLD"}],"text":"Queues — это очереди задач, которые ждут выполнения.\nПример: На сайте 100 человек одновременно хотят купить билет. Если сервер может обработать только 10 заказов в секунду, остальные 90 попадают в очередь.","type":"unordered-list-item","key":"2o06t"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"},{"length":7,"offset":83,"style":"BOLD"}],"text":"Latency — это время, которое проходит между отправкой запроса и получением ответа.\nПример: Вы нажали на кнопку «Купить», а ответ пришёл через 2 секунды — это и есть latency.","type":"unordered-list-item","key":"68ocl"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":4,"offset":0,"style":"BOLD"},{"length":7,"offset":63,"style":"BOLD"}],"text":"inc_* — функции, которые увеличивают счётчики (counters) на 1.\nПример: inc_orders() — увеличивает счётчик заказов на 1 каждый раз, когда кто-то делает заказ.","type":"unordered-list-item","key":"b01tp"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":4,"offset":0,"style":"BOLD"},{"length":7,"offset":68,"style":"BOLD"}],"text":"set_* — функции, которые устанавливают текущее значение для gauges.\nПример: set_queue_length(5) — говорит, что в очереди сейчас 5 задач.","type":"unordered-list-item","key":"6uo5n"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":20,"offset":0,"style":"BOLD"}],"text":"Реал-тайм мониторинг — это наблюдение за метриками прямо сейчас, без задержек.","type":"unordered-list-item","key":"990d5"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":11,"offset":0,"style":"BOLD"},{"length":7,"offset":123,"style":"BOLD"}],"text":"Autoscaling — автоматическое увеличение или уменьшение количества серверов (подов в Kubernetes) в зависимости от нагрузки.\nПример: Если очередь задач стала слишком большой, Kubernetes автоматически запустит ещё один сервер, чтобы справиться с нагрузкой.","type":"unordered-list-item","key":"lu2s"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"29al1"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"6. Main — DI и запуск","type":"header-three","key":"flc07"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Инициализация tracing + metrics.","type":"unordered-list-item","key":"7se0a"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Создание адаптеров (reader, processor, enricher с GPU, writers transactional, checkpoint).","type":"unordered-list-item","key":"asqp3"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Сборка orchestrator с конфигурацией parallelism/batch_size/threshold.","type":"unordered-list-item","key":"4t393"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"orchestrator.run().await — запуск пайплайна.","type":"unordered-list-item","key":"711tv"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Подробное объяснение ML-модели в Enricher (максимальная архитектурная сложность)","type":"header-three","key":"dmtg1"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":10,"offset":0,"style":"BOLD"},{"length":52,"offset":14,"style":"BOLD"},{"length":5,"offset":141,"style":"BOLD"}],"text":"MlEnricher — многопоточный параллельный batch-инференс на GPU/CPU с использованием PyTorch (через tch-rs bindings) строго следует принципам SOLID:","type":"unstyled","key":"bhssi"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":3,"offset":0,"style":"BOLD"}],"text":"SRP: только обогащение батча изменений ML-предсказаниями (anomaly detection + classification).","type":"unordered-list-item","key":"90id9"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":3,"offset":0,"style":"BOLD"}],"text":"OCP: расширяем (новые модели/фичи добавляются без изменения кода).","type":"unordered-list-item","key":"7pm8q"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":3,"offset":0,"style":"BOLD"}],"text":"LSP: реализует ChangeEnricher трейт — взаимозаменяем с другими enricher'ами.","type":"unordered-list-item","key":"11j19"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":3,"offset":0,"style":"BOLD"}],"text":"ISP: узкий интерфейс — только enrich_batch.","type":"unordered-list-item","key":"q198"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":3,"offset":0,"style":"BOLD"}],"text":"DIP: зависит от абстракции TorchModelProvider (загрузка модели).","type":"unordered-list-item","key":"4diue"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":30,"offset":4,"style":"BOLD"}],"text":"это production-grade ML-интеграция в real-time поток с backpressure, latency-метриками и fallback на CPU.","type":"unstyled","key":"4fn2v"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"1. TorchModelProvider — провайдер модели (одноразовая загрузка)","type":"header-three","key":"4ckvs"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"i9gj"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":491,"offset":0,"style":"BOLD"}],"text":"pub struct TorchModelProvider {\n model: CModule, // TorchScript модуль (экспортированная модель)\n device: Device, // GPU (Cuda) если доступно, иначе CPU\n}\n\nimpl TorchModelProvider {\n pub fn new(path: &str, device: Device) -> Result {\n let mut model = CModule::load(path)?; // Загрузка .pt (TorchScript) один раз\n model.set_eval(); // Режим inference (no dropout, batchnorm fixed)\n Ok(Self { model, device })\n }\n}","type":"unstyled","key":"djc6s"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":12,"offset":0,"style":"BOLD"},{"length":8,"offset":33,"style":"BOLD"}],"text":"Как работает: Модель загружается один раз при старте (lazy initialization в main через Arc для шаринга между потоками).\nCModule::load — читает TorchScript файл (.pt), экспортированный из PyTorch (torch.jit.script или trace).\nset_eval() — отключает training-режим для детерминированного инференса.\ndevice: Device::cuda_if_available() — автоматический GPU-fallback (максимальная производительность: тысячи событий/сек на modern GPU).\n","type":"unordered-list-item","key":"4te89"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Поддержка GPU без дополнительного кода (tch-rs использует LibTorch с CUDA).\nМодель может быть любой: Autoencoder для anomaly reconstruction error, LSTM для последовательных паттернов, Transformer для контекста, или ensemble.\n","type":"unordered-list-item","key":"79gkq"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Пример модели (как обучить в PyTorch и экспортировать):","type":"unstyled","key":"8gsap"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":554,"offset":0,"style":"BOLD"}],"text":"class AnomalyClassifier(torch.nn.Module):\n def __init__(self, input_dim=10):\n super().__init__()\n self.fc = torch.nn.Sequential(\n torch.nn.Linear(input_dim, 128),\n torch.nn.ReLU(),\n torch.nn.Linear(128, 64),\n torch.nn.ReLU(),\n torch.nn.Linear(64, 2) # [anomaly_score (0-1), class_prob (critical)]\n )\n \n def forward(self, x):\n return torch.sigmoid(self.fc(x)) # или raw logits\n\n# Обучение...\nscripted = torch.jit.script(model.eval())\nscripted.save(\"anomaly.pt\")","type":"unstyled","key":"8uva0"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"2. MlEnricher — батчевый инференс с feature extraction","type":"header-three","key":"dcvs2"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"6qcsd"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":2321,"offset":0,"style":"BOLD"}],"text":"pub struct MlEnricher {\n model: Arc, // Шаринг между потоками (thread-safe)\n}\n\nimpl MlEnricher {\n pub fn new(model: Arc) -> Self { Self { model } }\n\n fn extract_features(&self, change: &ProcessedChange) -> Vec {\n match change {\n ProcessedChange::Order(o) => o.synthetic_features.clone(), // Из transformer\n _ => vec![0.0; 10], // Fallback для неизвестных типов (OCP)\n }\n }\n}\n\n#[async_trait]\nimpl ChangeEnricher for MlEnricher {\n async fn enrich_batch(&self, batch: Vec) -> Result, CdcError> {\n if batch.is_empty() { return Ok(vec![]); }\n\n // 1. Feature extraction (параллельно по батчу)\n let features: Vec> = batch.iter().map(|c| self.extract_features(c)).collect();\n\n // 2. Батчинг в Tensor (максимальная эффективность GPU)\n let tensors: Vec = features.iter()\n .map(|v| Tensor::from_slice(v).to(self.model.device)) // На GPU!\n .collect();\n let input = Tensor::stack(&tensors, 0); // [batch_size, feature_dim]\n\n // 3. Parallel batch inference (весь батч за один forward!)\n let output = self.model.model.forward_ts(&[input])?; // GPU-accelerated\n let cpu_output = output.to_device(Device::Cpu); // Только для чтения результатов\n let values: Vec = cpu_output.f_values()?; // Flat vector результатов\n\n // 4. Парсинг output (предполагаем модель возвращает [anomaly_score, class_prob] на строку)\n let feature_dim = values.len() / batch.len(); // Динамический (поддержка разных моделей)\n let chunks = values.chunks(feature_dim);\n\n let mut enriched = Vec::with_capacity(batch.len());\n for (i, chunk) in chunks.enumerate() {\n let anomaly_score = chunk[0]; // 0-1 sigmoid\n let classification = if chunk.get(1).map(|&p| p > 0.5).unwrap_or(false) {\n \"critical\".to_string()\n } else {\n \"normal\".to_string()\n };\n\n enriched.push(EnrichedChange {\n inner: batch[i].clone(),\n anomaly_score: Some(anomaly_score),\n classification: Some(classification),\n });\n }\n\n Ok(enriched)\n }\n}","type":"unstyled","key":"3ahks"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Как работает inference шаг за шагом (максимальная сложность)","type":"header-three","key":"8hq42"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":5,"offset":0,"style":"BOLD"}],"text":"Input: батч ProcessedChange (из transformer: synthetic_features — нормализованные признаки вроде log(amount), user_id, timestamp).","type":"ordered-list-item","key":"226b6"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":18,"offset":0,"style":"BOLD"}],"text":"Feature extraction: O(1) per event — простая копия/генерация (расширяемо для complex embedding).","type":"ordered-list-item","key":"5cqfj"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":7,"offset":0,"style":"BOLD"}],"text":"Батчинг:Каждый вектор фич → Tensor на device (GPU memcpy — асинхронно).\nTensor::stack — один большой тензор [batch_size, dim] (256–1024 для оптимальной GPU utilization).\n","type":"ordered-list-item","key":"965i4"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":12,"offset":0,"style":"BOLD"}],"text":"Forward pass:forward_ts(&[input]) — один вызов модели на весь батч (параллелизм CUDA kernels: тысячи операций одновременно).\nLatency ~1-10ms на modern GPU для батча 256 (vs секунды на sequential CPU).\n","type":"ordered-list-item","key":"dqnfj"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":14,"offset":0,"style":"BOLD"}],"text":"Output parsing:to_cpu + f_values — эффективное чтение.\nДинамический chunking — модель может возвращать любой вектор (OCP).\nanomaly_score: reconstruction error или probability.\nclassification: threshold на prob (critical если >0.5).\n","type":"ordered-list-item","key":"4l9b7"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"Почему это максимально эффективно?","type":"header-three","key":"216cv"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":15,"offset":0,"style":"BOLD"}],"text":"Многопоточность: enrich_batch вызывается параллельно в orchestrator (parallelism_per_stage потоков) — динамический батчинг (orchestrator накапливает до batch_size).","type":"unordered-list-item","key":"8bblg"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":16,"offset":0,"style":"BOLD"}],"text":"GPU acceleration: весь батч на CUDA — 10-100x speedup vs CPU loop.","type":"unordered-list-item","key":"5ttth"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":24,"offset":0,"style":"BOLD"}],"text":"Backpressure integration: если GPU overloaded — очередь растёт, orchestrator замедляет readers.","type":"unordered-list-item","key":"2uvlb"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":13,"offset":0,"style":"BOLD"}],"text":"Observability: latency метрика в Prometheus (record_inference_latency в orchestrator).","type":"unordered-list-item","key":"f7fiu"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":13,"offset":0,"style":"BOLD"}],"text":"Расширяемость: новая модель (Siamese для dedup, LSTM для sequences) — просто новый .pt файл.","type":"unordered-list-item","key":"fcevt"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[{"length":10,"offset":0,"style":"BOLD"}],"text":"Надёжность: TchError → CdcError, fallback CPU.","type":"unordered-list-item","key":"buocs"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"В orchestrator это интегрировано с роутингом: если anomaly_score >= threshold — в quarantine Kafka topic (интеллектуальная фильтрация подозрительных изменений в реал-тайм).","type":"unstyled","key":"3u2rm"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"6nlb7"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"3f0me"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"unstyled","key":"d0v6g"},{"data":{},"depth":0,"entityRanges":[],"inlineStyleRanges":[],"text":"","type":"marker","key":"IAwqlC5voQ8iW8PtcmYv"}],"entityMap":{}},"mentions":[]}