Оглавление

1. 1. Резюме проекта
2. 2. Описание задачи и целей автоматизации
3. 3. Архитектура решения
4. 4. Описание сценариев автоматизации
5. 5. Структура данных в Supabase
6. 6. Механизмы безопасности
7. 7. Механизмы надежности
8. 8. Интеграция с AI (GigaChat)
9. 9. Результаты тестирования
10. 10. Отчетность и визуализации
11. 11. Описание интеграций
12. 12. Чек-лист соответствия ТЗ
13. Медиа-материалы кейса

1. Резюме проекта

Разработан и внедрен полностью автоматизированный конвейер обработки и анализа больших данных на платформе n8n 2.36.9 (Self-Hosted) с использованием Supabase в качестве хранилища данных и GigaChat для генерации аналитических инсайтов.

Конвейер работает по архитектуре IN_ → CORE_ → OUT_ → UTIL_ и включает четыре основных сценария:

• IN_DataInsight_Ingest_Sources — загрузка данных из двух внешних источников (JSONPlaceholder и DummyJSON) с масштабированием во временной ряд.
• CORE_DataInsight_Clean_Aggregate — очистка, валидация, дедупликация и агрегация данных.
• CORE_DataInsight_AI_Analyze — статистический анализ, выявление аномалий, построение прогноза и генерация AI-инсайтов через GigaChat.
• OUT_DataInsight_Report_Notifier — генерация HTML-отчета с SVG-графиками и отправка уведомления в Telegram.

Ключевые показатели проекта

2. Описание задачи и целей автоматизации

Заказчик

DataInsight — крупная аналитическая фирма, специализирующаяся на обработке и анализе больших данных для отраслей финансов, здравоохранения, ритейла и технологий.

Боли, которые решены

• ✅ Ручная обработка больших объемов данных — заменена автоматизированным конвейером.
• ✅ Отсутствие системы очистки и агрегации из различных источников — реализован единый пайплайн с валидацией и дедупликацией.
• ✅ Множество ручных процессов извлечения и анализа — полностью автоматизировано.
• ✅ Отсутствие интеграции между системами — все сценарии работают как единый конвейер через вебхуки и общую базу данных.

Цели автоматизации

• ✅ Автоматизация процесса очистки, агрегации и анализа данных с использованием Big Data технологий.
• ✅ Интеграция с различными источниками данных (внешние API) для быстрой обработки больших объемов информации.
• ✅ Ускорение получения аналитических отчетов и прогнозных моделей (время полного цикла — единицы секунд).
• ✅ Уменьшение времени обработки данных и повышение точности аналитики (статистические методы + AI).
• ✅ Автоматизация генерации отчетов и визуализаций для принятия обоснованных решений.

3. Архитектура решения

Общая схема конвейера данных

Технологический стек

4. Описание сценариев автоматизации

4.1 IN_DataInsight_Ingest_Sources — загрузка данных

Триггеры: вебхук POST /webhook/datain-ingest и расписание каждые 6 часов.

Назначение: забор данных из двух внешних источников, нормализация к единой схеме и передача в пайплайн очистки.

Общая схема сценария

Webhook, Schedule, Build Pages, Fetch, Normalize, Scale, Chunk, Send

Обработанный объем за тест: 6 036 записей, 3 батча.

4.2 CORE_DataInsight_Clean_Aggregate — очистка и агрегация

Триггер: вебхук POST /webhook/datain-clean.

Общая схема сценария

Clean → IF → Dedup → Upsert → Recalc

4.3 CORE_DataInsight_AI_Analyze — аналитика

Триггеры: вебхук POST /webhook/datain-analyze и расписание ежедневно в 07:00.

Общая схема сценария

Fetch → Analyze → AI Agent → GigaChat → Extract → Save

4.4 OUT_DataInsight_Report_Notifier — отчеты и уведомления

Триггеры: вебхук POST /webhook/datain-report и расписание ежедневно в 08:00.

Общая схема сценария

Load → Generate → Save → Telegram → Respond

4.5 Global Error Handler — обработка ошибок

Сценарий подключен ко всем основным сценариям через параметр errorWorkflow.

Общая схема сценария

Format → Analyze → Increment → Wait → Restart → Alert

4.6 IN_ExternalUsers_Fetch — экспериментальный сценарий (Proof of Concept)

Триггеры: вебхук и расписание каждые 8 часов. Статус: отключен (active: false).

Назначение: полигон для отработки базовых механизмов конвейера и демонстрация адаптерного паттерна подключения внешних источников. Загружает пользователей из jsonplaceholder.typicode.com/users — источник, содержащий персональные данные (name, email).

Общая схема сценария

Webhook, Schedule, Fetch Page, Normalize (Adapter), Validate Data, DLQ

Механизмы, отработанные на полигоне и внедренные в продакшн

• ✓ Идемпотентность и дедупликация (повторный вызов не создает дублей);
• ✓ Валидация данных и Dead Letter Queue;
• ✓ Адаптерный паттерн подключения новых источников;
• ✓ Механизм повторных запусков через retry_session_id;
• ✓ Единая схема передачи батчей в /webhook/datain-clean.

4.7 Retry Router (Global) — маршрутизатор перезапусков

Триггер: вебхук POST /webhook/retry-trigger. Статус: активен.

Назначение: принимает запрос на перезапуск от Global Error Handler (узел Restart Workflow), восстанавливает контекст повторной попытки и вызывает вебхук исходного сценария. Замыкает цикл ретраев: ошибка → классификация → счетчик попыток → перезапуск → при повторной ошибке контекст сохраняется и цепочка продолжается.

Общая схема сценария

retry-trigger → контекст → Build JSON Body → HTTP Request на вебхук цели

Механизмы защиты цикла ретраев

• ✓ Не более 3 попыток — счетчик ведет атомарная функция increment_retry_session() на стороне Global Error Handler;
• ✓ 5-минутное окно группировки ошибок — дубли не создают параллельных цепочек;
• ✓ Контекст retry_session_id передается в новый запуск — при повторной ошибке цепочка продолжается, а не начинается заново;
• ✓ neverError в HTTP-вызове — сбой перезапуска не порождает новый цикл ошибок;
• ✓ Маршрутизация по original_workflow_id — перезапускается именно тот сценарий, в котором произошла ошибка.

5. Структура данных в Supabase

Ключевая функция корректной агрегации:

-- Пересчет агрегатов по всем записям за указанные даты
-- Вызывается после каждого батча через RPC
SELECT public.recalc_daily_aggregates(ARRAY['2026-09-04']::date[]);

Функция идемпотентна и защищает от состояния гонки: даже если несколько батчей за один день приходят одновременно, агрегаты будут корректно пересчитаны по всем записям, а не перезаписаны последним батчем.

6. Механизмы безопасности

7. Механизмы надежности

7.1 Идемпотентность (защита от дублей)

Все операции записи спроектированы как идемпотентные:

• di_clean_records — UPSERT по уникальной паре (source, dedup_hash), где dedup_hash вычисляется алгоритмом FNV-1a из source|record_id.
• daily_aggregates — пересчет через функцию recalc_daily_aggregates() с ON CONFLICT (agg_date, category) DO UPDATE.
• analysis_results — UPSERT с on_conflict=run_date.
• reports — UPSERT с on_conflict=report_date.

Подтверждено тестом: повторный запуск datain-clean и datain-analyze не создает дублей (cnt = 1 в analysis_results и reports за текущую дату).

7.2 Дедупликация внутри батча

Узел Dedup In Batch использует Set для отслеживания уже встреченных dedup_hash и отбрасывает дубли до записи в БД.

7.3 Обработка временных сбоев и защита от бесконечных повторов

• Для критических записей в БД заданы таймауты (30 секунд) и ограниченное число повторных попыток.
• Global Error Handler использует атомарную функцию increment_retry_session() с лимитом 3 попытки и 5-минутным окном группировки ошибок.
• Результат обработки фиксируется в таблице workflow_errors независимо от исхода.
• При исчерпании попыток отправляется финальный алерт в Telegram.

7.4 Защита от состояния гонки

Функция increment_retry_session() использует pg_advisory_xact_lock() для сериализации параллельных вызовов. Пересчет агрегатов через recalc_daily_aggregates() выполняется на стороне БД, что исключает конфликтующие записи при одновременной обработке нескольких батчей.

7.5 Graceful degradation

• Сбой отправки уведомления в Telegram логируется, но не прерывает пайплайн.
• Недоступность AI переключает систему на эвристические инсайты.
• Невалидные записи не блокируют батч, а сохраняются в Dead Letter Queue.

8. Интеграция с AI (GigaChat)

AI реализован как сервис внутри процесса: он возвращает строго заданный результат, а сценарий его проверяет, принимает решение и управляет дальнейшими действиями.

8.1 Контракт взаимодействия

В узле AI Agent задан системный промпт, требующий ответ строго в формате JSON:

{
  "summary": "краткий вывод",
  "bullets": ["инсайт 1", "инсайт 2"],
  "risks": ["риск 1", "риск 2"],
  "trend": "up|down|flat"
}

8.2 Валидация ответа

Узел Extract AI Text выполняет проверку:

• Ответ не пустой.
• Присутствуют обязательные поля summary и bullets.
• Если JSON обернут в текст — извлекается через регулярное выражение.

При невалидном ответе или ошибке система автоматически переключается на эвристические инсайты (узел Heuristic Insights).

8.3 Пример сгенерированных инсайтов

Выявлены признаки снижения ключевых метрик с августа; зафиксированы аномальные всплески продаж.

Ключевые инсайты:
- Снижение среднего дневного объема продаж (-23 796)
- Аномальный рост продаж 9 августа 2026 года до 38,9 млн ($), 23 августа — до 41,3 млн ($)

Риски:
- Продолжение нисходящего тренда может свидетельствовать о снижении спроса
- Высокие отклонения от средних значений требуют дополнительного анализа причин

9. Результаты тестирования

Executions — список

Mode=webhook, Status=Success

Загрузка данных

Анализ данных

Генерация отчета

9.1 Сквозной тест пайплайна (04.09.2026)

9.2 Объемы данных

9.3 Тесты идемпотентности

9.4 Тесты обработки ошибок

10. Отчетность и визуализации

10.1 Структура автоматического отчета

HTML-отчет генерируется узлом Generate Report и содержит:

• Ключевые показатели (KPI): всего записей, дней в выборке, среднее в день, количество аномалий, направление тренда.
• Бар-чарт по категориям: топ-10 категорий по сумме (инлайн SVG).
• Линейный график динамики: история за 31 день + прогноз на 7 дней пунктиром (инлайн SVG).
• Таблица аномалий: даты с отклонением |z| > 2.5.
• AI-инсайты: сгенерированный GigaChat анализ.

KPI-блоки отчета, Бар-чарт топ-категорий

6 036 записей, 31 день, тренд вниз, 2 аномалии, vehicle, mens-watches, womens-watches и т.д.

Линейный график с прогнозом. Блок AI-инсайтов

Синяя линия — история, оранжевый пунктир — прогноз. GigaChat анализ с инсайтами и рисками

10.2 Пример сводки из Telegram

📊 DataInsight: отчет за 04.09.2026
Записей: 6 036 · Дней: 31 · Среднее/день: 20 789 721,43
Тренд: спад · Аномалий: 2
Топ-категория: vehicle (417 799 737,93)

🤖 Средние ежедневные продажи снизились за отчетный период, выявлены две аномальные даты с высокими значениями.

Ежедневный отчет в Telegram

Сводка с KPI и AI-инсайтами

10.3 Хранение отчетов

Каждый отчет сохраняется в таблицу reports с полями report_date, title, html, summary. Идемпотентность гарантирована уникальным ограничением на report_date.

11. Описание интеграций

11.1 Источники данных

11.2 Аналитические инструменты

11.3 Хранилище данных

11.4 Каналы уведомлений

12. Чек-лист соответствия ТЗ

Сдаваемые артефакты

Медиа-материалы кейса

Все ключевые артефакты проекта. Нажмите на любой скриншот для увеличения в модальном окне.

Видео: демонстрация работы пайплайна

Ежедневный отчет в Telegram

Алерт о критической ошибке

Выполнение IN_Ingest

5820 записей загружено, 3 батча

Выполнение CORE_Clean

Очистка и агрегация батча

Выполнение CORE_Analyze

Анализ + AI-инсайты от GigaChat

Выполнение OUT_Report

Генерация HTML + Telegram

IN_ExternalUsers_Fetch (полигон)

Демо адаптера и валидации; статус — отключен из соображений защиты ПД