Решение "Массовая обработка данных"

Назначение

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

Описание

Шаблон разработки для массовой обработки данных в Directum RX
Статья на Directum Club

Возможности решения

Обеспечение стабильного, контролируемого и эффективного процесса обработки больших объемов данных (миллионы сущностей) с минимальными трудозатратами на разработку и защитой производительности системы.

Состав решения

1. Абстрактный справочник-конфигуратор с настройками потоков, размера пачек, повторных попыток и лимитов потоков на рабочее и не рабочее время.
2. Справочник элемента очереди для хранения списка обрабатываемых объектов и информации о результатах обработки.
3. Фоновый процесс для создания очередей и запуска асинхронных обработчиков с учетом заданных лимитов.
4. Асинхронный обработчик, устойчивый к ошибкам: ошибки изолируются, фиксируются и не приводят к неконтролируемому росту числа процессов.
5. Набор клиентских методов, предназначенных для добавления действий на обложку модуля.
6. Не содержит перекрытий «коробочных» сущностей.

Порядок установки

Для работы требуется установленный Directum Development studio версии 4.10 и выше

Установка для ознакомления

1. Склонировать репозиторий BulkProcessing в папку (например C:\WorkFolder).
2. Указать в _ConfigSettings.xml DDS:

<block name="REPOSITORIES">
  <repository folderName="Base" solutionType="Base" url="<адрес локального репозитория>" />
  <repository folderName="Work" solutionType="Work" url="<адрес локального репозитория>" />
  <repository folderName="<Папка из п.1>" solutionType="Work" url="https://github.com/STARKOV-Group/BulkProcessing.git" />
</block>

Установка для использования на проекте

Возможные варианты:

A. Fork репозитория

1. Сделать fork репозитория BulkProcessing для своей учетной записи.
2. Склонировать созданный в п. 1 репозиторий в папку.
3. Указать в _ConfigSettings.xml DDS:

<block name="REPOSITORIES">
  <repository folderName="Base" solutionType="Base" url="<адрес локального репозитория>" />
  <repository folderName="Work" solutionType="Work" url="<адрес локального репозитория>" />
  <repository folderName="<Папка из п.2>" solutionType="Work" url="<Адрес репозитория gitHub учетной записи пользователя из п. 1>" />
</block>

B. Подключение на базовый слой.

Вариант не рекомендуется:

• так как при выходе новой версии шаблона разработки не гарантируется обратная совместимость;
• потеряется возможность изменения или доработки функционала под собственые требования;

1. Склонировать репозиторий BulkProcessing в папку.
2. Указать в config.yml DirectumLauncher:

			-   '@folderName': 'BulkProcessing'
                '@solutionType': 'Work'
                '@url': 'https://github.com/STARKOV-Group/BulkProcessing.git'

C. Копирование репозитория в систему контроля версий.

Рекомендуемый вариант для проектов внедрения.

1. В системе контроля версий с поддержкой git создать новый репозиторий.
2. Склонировать репозиторий BulkProcessing в папку с ключом --mirror.
3. Перейти в папку из п. 2.
4. Импортировать клонированный репозиторий в систему контроля версий командой:

git push –mirror <Адрес репозитория из п. 1>

Создание нового процесса

Для создания нового обработчика, на рабочем слое своего решения требуется создать справочник-наследник от абстрактного предка из шаблона ProcessSettingBase («Настройка процесса»). При необходимости можно расширить форму дополнительными свойствами. Далее потребуется переопределить несколько методов, необходимых для кастомизации логики:

    /// <summary>
    /// Получить все сущности.
    /// </summary>
    public override IQueryable<Sungero.Domain.Shared.IEntity> GetAllEntities()
    {
      // Вернуть результат метода GetAll() типа сущности, который будет в обработке.
    }

    /// <summary>
    /// Получить сущности для обработки.
    /// </summary>
    public override IQueryable<Sungero.Domain.Shared.IEntity> GetEntitiesForProcessing()
    {
      // Ваша логика получения объектов (наследников IEntity), которые требуется обработать
    }

Опционально, вместо метода GetEntitiesForProcessing можно перекрыть метод GetEntitiesIdsForProcessing, возвращающий IQueryable, если получение идентификаторов сущностей предполагается не из репозитория.

    /// <summary>
    /// Обработать сущность.
    /// </summary>
    /// <param name="entity">Сущность для обработки.</param>
    /// <param name="_logger">Преднастроенный экземпляр логгера с контекстом (settingId, processId).</param>
    public override void ProcessEntity(Sungero.Domain.Shared.IEntity entity, Sungero.Core.ILogger _logger)
    {
      // Ваша логика обработки, которая должна применятся к каждому объекту
    }

Архитектурно неочевидные моменты

1. Решение может работать только с сущностями – наследниками IEntity.
2. В очереди хранятся идентификаторы объектов.
3. Для разработки нового процесса требуется наследование от абстрактного справочника ProcessSettingsBase с обязательным переопределением группы методов (сгруппированы в отдельный регион на серверном слое).

Note

Замечания и пожелания по развитию шаблона разработки фиксируйте через Issues. При оформлении ошибки, опишите сценарий для воспроизведения. Для пожеланий приведите обоснование для описываемых изменений - частоту использования, бизнес-ценность, риски и/или эффект от реализации.
Внимание! Изменения будут вноситься только в новые версии.