Редактирование

Markdown (tb.md)

# Airflow DAG: Создание кроссов на основе фида PCM

> **Процесс уровня:** L2 (Детальный процесс)
> **Категория:** Operational
> **Методология:** TO-BE Only
> **Способ проектирования:** Collaborative
> **Дата создания:** 2023-10-27
> **Статус:** Draft
---

## Цель процесса

Автоматическое создание и обновление кроссов запчастей и жидкостей в Pimcore на основе данных из фида PCM.

---

## Глоссарий и термины

| Термин | Определение |
|--------|-------------|
| **DAG** | Directed Acyclic Graph - направленный ациклический граф, используемый в Apache Airflow для определения последовательности задач. |
| **PCM** | Product Catalog Management - система управления продуктовым предложением, отображаемым клиентам. |
| **Pimcore** | Система управления данными о продуктах (PIM) и контентом (CMS). |
| **Фид PCM** | Файл или поток данных, содержащий информацию о запчастях и жидкостях из системы PCM. |
| **Кросс** | Связь между двумя запчастями/жидкостями, указывающая на их взаимозаменяемость или аналогичность. |
| **article** | Артикул запчасти/жидкости. |
| **original_article** | Оригинальный артикул запчасти/жидкости. |
| **brand** | Бренд запчасти/жидкости. |
| **MDM** | Master Data Management - система управления основными данными. |
| **clean/raw** | Варианты артикулов: "clean" - нормализованный, "raw" - исходный. |
| **GraphQL** | Язык запросов для API. |
| **Синоним бренда** | Альтернативное название бренда. |
| **MongoDB** | Документоориентированная система управления базами данных. |
| **RabbitMQ** | Брокер сообщений, используемый для асинхронной обработки данных. |

---

## Роли процесса

- **Команда PCM** — отвечает за сбор и предоставление фида PCM.
- **Разработчик DAG** — отвечает за реализацию и поддержку DAG Airflow.
- **Pimcore** (система) — система управления продуктовыми данными.
- **ClickHouse** (система) — аналитическая база данных, куда записываются данные фида PCM.
- **Airflow** (система) — платформа для программного создания, планирования и мониторинга рабочих процессов.
- **MongoDB** (система) — база данных для хранения кроссов.
- **RabbitMQ** (система) — брокер сообщений для обмена данными.

---

## Шаги процесса

### БЛОК 1: Инициация и получение данных

### `Step-1`: Запуск DAG Airflow
- **Тип:** serviceTask
- **Ответственный:** Airflow
- **Вход:** —
- **Trigger:** Ежедневное расписание (2 раза в день)
- **Описание:** Airflow запускает DAG по расписанию.
- **Действия:**
  1. Инициализация выполнения DAG.
- **Выход:** Запущенный процесс.
- **Система:** Airflow
- **Автоматизация:** Полная
- **Время:** Мгновенно
- **Следующие шаги:** `Step-2`

### `Step-2`: Ожидание завершения сбора фида PCM
- **Тип:** serviceTask
- **Ответственный:** Airflow
- **Вход:** —
- **Trigger:** Запуск DAG.
- **Описание:** DAG ожидает подтверждения завершения сбора фида PCM от команды PCM.
- **Действия:**
  1. Ожидание сигнала/условия от команды PCM.
- **Выход:** Подтверждение завершения сбора фида.
- **Система:** Airflow
- **Автоматизация:** Полная (по условию)
- **Время:** Зависит от команды PCM
- **Следующие шаги:** `Step-3`

### `Step-3`: Скачивание фида PCM / Доступ к ClickHouse
- **Тип:** serviceTask
- **Ответственный:** Airflow
- **Вход:** Ссылка на фид PCM или доступ к БД ClickHouse.
- **Trigger:** Завершение сбора фида PCM.
- **Описание:** DAG скачивает фид PCM по предоставленной ссылке или получает доступ к данным фида из ClickHouse.
- **Действия:**
  1. Выполнение HTTP-запроса для скачивания фида ИЛИ
  2. Выполнение запроса к ClickHouse для получения данных фида.
- **Выход:** Сырые данные фида PCM.
- **Система:** Airflow, ClickHouse
- **Автоматизация:** Полная
- **Время:** 5-30 мин
- **Следующие шаги:** `Step-4`

### БЛОК 2: Нормализация и получение данных из Pimcore

### `Step-4`: Нормализация полей фида PCM
- **Тип:** serviceTask
- **Ответственный:** Airflow
- **Вход:** Сырые данные фида PCM.
- **Trigger:** Получение данных фида PCM.
- **Описание:** Нормализация полей `article`, `original_article`, `brand` в данных фида PCM.
- **Действия:**
  1. Для `brand`: приведение к одному регистру, соответствующему брендам Pimcore.
  2. Для `article`, `original_article`: подготовка вариантов `clean` и `raw` (например, удаление дефисов).
- **Выход:** Нормализованные данные фида PCM.
- **Система:** Airflow
- **Автоматизация:** Полная
- **Время:** 10-60 мин
- **Следующие шаги:** `Step-5`

### `Step-5`: Запрос текущих запчастей и жидкостей из Pimcore
- **Тип:** serviceTask
- **Ответственный:** Airflow
- **Вход:** —
- **Trigger:** Нормализация данных фида PCM.
- **Описание:** Выполнение GraphQL-запроса к Pimcore для получения текущих запчастей и жидкостей с их кроссами.
- **Действия:**
  1. Формирование GraphQL-запроса.
  2. Отправка запроса к Pimcore.
  3. Получение ответа с данными.
- **Выход:** Список запчастей/жидкостей из Pimcore с их кроссами.
- **Система:** Airflow, Pimcore
- **Автоматизация:** Полная
- **Время:** 5-30 мин
- **Следующие шаги:** `Step-6`

### БЛОК 3: Поиск и дедупликация

### `Step-6`: Поиск запчастей/жидкостей из Pimcore в фиде PCM
- **Тип:** serviceTask
- **Ответственный:** Airflow
- **Вход:** Нормализованные данные фида PCM; список запчастей/жидкостей из Pimcore.
- **Trigger:** Получение данных из Pimcore.
- **Описание:** Для каждой запчасти/жидкости (clean) из Pimcore выполняется поиск в нормализованном фиде PCM.
- **Действия:**
  1. Поиск по `brand` и `article` (из Pimcore) в полях `article`, `brand` фида PCM.
  2. Поиск по синонимам `brand` и `article` (из Pimcore) в полях `article`, `brand` фида PCM.
  3. Дополнительная проверка на наличие запчасти в фиде в двух вариантах (с нормализацией и без) для `article` (например, TF0041300 и TF004-1300).
- **Выход:** Найденные соответствия в фиде PCM или лог о ненахождении.
- **Система:** Airflow
- **Автоматизация:** Полная
- **Время:** 30-120 мин
- **Следующие шаги:** `GW-1`

#### Gateway GW-1: Запчасть/жидкость найдена в фиде PCM?
- **Тип:** exclusiveGateway
- **ДА** (найдены соответствия в фиде PCM) → `Step-7`
- **НЕТ** (соответствия не найдены) → `Step-8`

### `Step-7`: Поиск по `original_article` и дедупликация
- **Тип:** serviceTask
- **Ответственный:** Airflow
- **Вход:** Найденные соответствия в фиде PCM.
- **Trigger:** Запчасть/жидкость найдена в фиде PCM.
- **Описание:** Проверка поля `original_article` и поиск уникальных значений, затем дедупликация результатов.
- **Действия:**
  1. Извлечение всех уникальных артикулов из поля `original_article`.
  2. Если `original_article` содержит только `null`, то `article` из фида считается оригинальным и ищется в `original_article`.
  3. Поиск этих уникальных значений по полю `original_article` в фиде.
  4. Дедупликация найденных пар по полям `article`, `original_article`, `brand`.
- **Выход:** Дедуплицированные пары для создания кроссов.
- **Система:** Airflow
- **Автоматизация:** Полная
- **Время:** 10-60 мин
- **Следующие шаги:** `Step-9`

### `Step-8`: Логирование ненахождения запчасти/жидкости
- **Тип:** serviceTask
- **Ответственный:** Airflow
- **Вход:** Информация о ненахождении запчасти/жидкости.
- **Trigger:** Запчасть/жидкость не найдена в фиде PCM.
- **Описание:** Запись в лог информации о запчастях/жидкостях из Pimcore, которые не были найдены в фиде PCM.
- **Действия:**
  1. Формирование лог-сообщения.
  2. Запись лог-сообщения.
- **Выход:** Лог-запись.
- **Система:** Airflow
- **Автоматизация:** Полная
- **Время:** Мгновенно
- **Следующие шаги:** `END-logged-not-found`

### БЛОК 4: Формирование и сохранение кроссов

### `Step-9`: Формирование пар для создания кроссов
- **Тип:** serviceTask
- **Ответственный:** Airflow
- **Вход:** Дедуплицированные пары для создания кроссов.
- **Trigger:** Дедупликация завершена.
- **Описание:** Определение типа кросса (двухсторонний, замена) на основе условий поиска.
- **Действия:**
  1. Если бренд запчасти найден без поиска синонимов и бренд совпадает с искомым брендом: создать двухсторонний кросс с типом "Замена".
  2. Если бренд запчасти найден при помощи поиска синонимов: ориентироваться на направление синонима для определения типа кросса.
- **Выход:** Сформированные пары кроссов с указанием типа.
- **Система:** Airflow
- **Автоматизация:** Полная
- **Время:** 5-30 мин
- **Следующие шаги:** `Step-10`

### `Step-10`: Создание/обновление элементов в MongoDB
- **Тип:** serviceTask
- **Ответственный:** Airflow
- **Вход:** Сформированные пары кроссов.
- **Trigger:** Формирование пар кроссов завершено.
- **Описание:** Запись или обновление информации о кроссах в коллекции `parts_crosses` в MongoDB.
- **Действия:**
  1. Подключение к MongoDB.
  2. Вставка или обновление документов в коллекции `parts_crosses`.
- **Выход:** Обновленные данные в MongoDB.
- **Система:** Airflow, MongoDB
- **Автоматизация:** Полная
- **Время:** 5-30 мин
- **Следующие шаги:** `Step-11`

### `Step-11`: Отправка результатов в RabbitMQ
- **Тип:** serviceTask
- **Ответственный:** Airflow
- **Вход:** Сформированные пары кроссов.
- **Trigger:** Создание/обновление элементов в MongoDB.
- **Описание:** Отправка данных о созданных/обновленных кроссах в очередь RabbitMQ для дальнейшей обработки и обновления в Pimcore.
- **Действия:**
  1. Подключение к RabbitMQ.
  2. Публикация сообщений в соответствующую очередь.
- **Выход:** Сообщения в очереди RabbitMQ.
- **Система:** Airflow, RabbitMQ
- **Автоматизация:** Полная
- **Время:** 1-10 мин
- **Следующие шаги:** `END-ready`

---

## Конечные события (End Events)

| ID | Маршрут | Результат |
|----|---------|-----------|
| `END-logged-not-found` | От `Step-8` | Запчасти/жидкости, не найденные в фиде PCM, залогированы. Процесс завершен для этих элементов. |
| `END-ready` | От `Step-11` | Кроссы созданы/обновлены в MongoDB и отправлены в RabbitMQ для обновления в Pimcore. |

---

## Бизнес-правила

| ID | Правило | Затронутые шаги |
|----|---------|-----------------|
| BR-1 | Фид PCM собирается 2 раза в день. | `Step-1` |
| BR-2 | Нормализация бренда включает приведение к одному регистру с брендами Pimcore. | `Step-4` |
| BR-3 | Нормализация артикулов включает подготовку вариантов clean/raw. | `Step-4` |
| BR-4 | Если `original_article` содержит только `null`, то `article` из фида считается оригинальным. | `Step-7` |
| BR-5 | Если бренд запчасти найден без поиска синонимов и бренд совпадает с искомым брендом, создается двухсторонний кросс типа "Замена". | `Step-9` |
| BR-6 | Если бренд запчасти найден при помощи поиска синонимов, тип кросса определяется направлением синонима. | `Step-9` |

---

## Mermaid-диаграмма процесса

```mermaid
graph TD
    A[Начало] --> Step-1;

subgraph Инициация и получение данных
        Step-1(Step-1: Запуск DAG Airflow) --> Step-2;
        Step-2(Step-2: Ожидание завершения сбора фида PCM) --> Step-3;
        Step-3(Step-3: Скачивание фида PCM / Доступ к ClickHouse) --> Step-4;
    end

subgraph Нормализация и получение данных из Pimcore
        Step-4(Step-4: Нормализация полей фида PCM) --> Step-5;
        Step-5(Step-5: Запрос текущих запчастей и жидкостей из Pimcore) --> Step-6;
    end

subgraph Поиск и дедупликация
        Step-6(Step-6: Поиск запчастей/жидкостей из Pimcore в фиде PCM) --> GW-1;
        GW-1{GW-1: Запчасть/жидкость найдена в фиде PCM?}
        GW-1 -- ДА --> Step-7;
        GW-1 -- НЕТ --> Step-8;
        Step-7(Step-7: Поиск по original_article и дедупликация) --> Step-9;
        Step-8(Step-8: Логирование ненахождения запчасти/жидкости) --> END-logged-not-found;
    end

subgraph Формирование и сохранение кроссов
        Step-9(Step-9: Формирование пар для создания кроссов) --> Step-10;
        Step-10(Step-10: Создание/обновление элементов в MongoDB) --> Step-11;
        Step-11(Step-11: Отправка результатов в RabbitMQ) --> END-ready;
    end

classDef endEvent fill:#FFF,stroke:#333,stroke-width:2px,color:#333;
    class END-logged-not-found endEvent;
    class END-ready endEvent;
```

---

## Роли и ответственность (RACI)

| Шаг | Responsible | Accountable | Consulted | Informed | Автоматизация |
|-----|------------|-------------|-----------|----------|---------------|
| Step-1 | Airflow | Разработчик DAG | — | — | Полная |
| Step-2 | Airflow | Разработчик DAG | Команда PCM | — | Полная |
| Step-3 | Airflow | Разработчик DAG | — | — | Полная |
| Step-4 | Airflow | Разработчик DAG | — | — | Полная |
| Step-5 | Airflow | Разработчик DAG | — | — | Полная |
| Step-6 | Airflow | Разработчик DAG | — | — | Полная |
| Step-7 | Airflow | Разработчик DAG | — | — | Полная |
| Step-8 | Airflow | Разработчик DAG | — | — | Полная |
| Step-9 | Airflow | Разработчик DAG | — | — | Полная |
| Step-10 | Airflow | Разработчик DAG | — | — | Полная |
| Step-11 | Airflow | Разработчик DAG | — | — | Полная |

---

## Системы и интеграции

| ID | Интеграция | Источник | Целевая система | Данные | Сложность |
|----|-----------|----------|----------------|--------|-----------|
| INT-1 | Получение фида PCM | Внешний источник (ссылка) / ClickHouse | Airflow | Данные о запчастях и жидкостях | Средняя |
| INT-2 | Запрос данных из Pimcore | Airflow | Pimcore | Текущие запчасти/жидкости и их кроссы | Средняя |
| INT-3 | Запись в MongoDB | Airflow | MongoDB | Сформированные кроссы | Средняя |
| INT-4 | Отправка в RabbitMQ | Airflow | RabbitMQ | Данные для обновления Pimcore | Средняя |

---

## Сводка

| Метрика | Значение |
|---------|----------|
| Всего шагов | 11 |
| Всего шлюзов | 1 |
| Конечных событий | 2 |

Mermaid Preview

Редактирование tb.md