Перейти к содержимому
Puzzle RPA

Создать процесс (Dag)

image_1

Описание

Блок предназначен для создания процесса (DAG), который загружается на сервер оркестратора Airflow и представляет собой набор инструкций выполняемых оркестратором.

Блок поддерживает оркестраторы Airflow 2 и Airflow 3, версия выбирается в выпадающем списке блока и определяет, какой код будет сгенерирован.

Описание параметров

Блок имеет ряд параметров:

  1. Версия - версия оркестратора Airflow, под которую генерируется код процесса:

    Тип данных: dropdown (Выпадающий список)

    Пример: Airflow 3

    • Airflow 2 - генерируется код, совместимый с Airflow 2.
    • Airflow 3 - используются импорты и состав задач Airflow 3.

  2. Наименование - уникальный идентификатор процесса, отображаемый в веб-интерфейсе оркестратора;

    Тип данных: str (Строка)

    Пример: puzzle_robot

  3. Описание - краткое описание процесса для отображения в интерфейсе оркестратора;

    Тип данных: str (Строка)

    Пример: Робот по созданию актов в 1С.

  4. Дата отсчета интервала - дата, с которой нужно начинать планирование запуска процесса;

    Тип данных: str (Строка)

    Пример: 01.01.2025

  5. Время запуска - интервалы запуска, указываемые в виде cron-выражения. Проверить или составить cron-выражения можно с помощью сервиса: https://crontab.guru/;

    Тип данных: str (Строка)

    Пример: */30 08-19 * * 1-5

  6. Одновременный запуск процессов - максимальное количество одновременных запусков процесса;

    Тип данных: int (Целое число)

    Пример: 3

  7. Теги - параметр для сортировки процессов через web-интерфейс оркестратора. Для одного процесса можно указать несколько тегов;

    Тип данных: list (Массив)

    Пример: ['Пример', '1С']

  8. Аргументы по умолчанию:

    • owner - владелец процесса;

      Тип данных: str (Строка)

      Пример: Puzzle RPA

    • depends_on_past - стоит ли инициировать запуск следующего процесса, если предыдущий процесс в цепочке операций завершился с ошибкой;

      Тип данных: bool (Логический тип)

      Пример: Истина

    • email - список адресов электронной почты, для уведомлений о статусе выполнения;

      Тип данных: list (Массив)

    • email_on_failure - нужно ли отправлять уведомления на адрес электронной почты в случае неудачного выполнения;

      Тип данных: bool (Логический тип)

      Пример: Истина

    • retries - количество попыток повторного выполнения процесса в случае его неудачного завершения;

      Тип данных: int (Целое число)

      Пример: 3

    • retry_delay - время задержки перед повторным выполнением робота в случае его неудачного завершения;

      Тип данных: int (Целое число)

      Пример: 30

    • pool - параметр для управления очередями между различными процессами оркестратора;

      Тип данных: str (Строка)

      Пример: default_pool

  9. Выполнять пропущенные процессы - если значение параметра установлено в True, то все задачи, которые должны были быть выполнены в прошлом, будут выполнены при запуске процесса. Если значение параметра установлено в False, то только текущие и будущие задачи будут выполнены.

    Тип данных: bool (Логический тип)

    Пример: Истина

  10. Параметры запуска - глобальные параметры по умолчанию;

    Тип данных: dict (Словарь)

    Пример: {'key1': 'value2'}

    В данном примере определяются глобальные параметры запуска DAG в виде словаря, где:

    • Ключ - имя параметра;
    • Значение - данные по умолчанию.
    image_4

    При запуске DAG из веб-интерфейса Airflow параметры отображаются в окне Запуск Dag-а в блоке Run Parameters - перед запуском их значения можно изменить:

    image_5

    Получить параметры роботом можно с помощью блока Использовать аргумент запуска.

  11. Имя пула - имя пула Airflow, который соответствует виртуальной машине для выполнения роботов. Пул занимается на всё время работы роботов процесса, поэтому роботы других процессов на эту машину не заходят;

    Тип данных: str (Строка)

    Пример: default_pool

    Управление очередями запуска Создание пулов в Airflow и ограничение одновременного выполнения роботов. Подробнее...

  12. Время ожидания пула, с - сколько секунд процесс готов простоять в очереди к виртуальной машине, пока слот пула занят другим процессом;

    Тип данных: int (Целое число)

    Пример: 600

  13. Максимальное время удержания пула, с - предохранитель: предельное время в секундах, в течение которого процесс удерживает слот пула. Дольше указанного времени слот не удерживается;

    Тип данных: int (Целое число)

    Пример: 7200

  14. Роботы - разъем, который определяет, какие роботы будут выполнятся процессом.

    Тип данных: Блоки функциональных действий студии разработки Puzzle RPA

    Допускается использование блоков:

  15. Порядок выполнения роботов - объявление зависимостей отдельных роботов.

    Тип данных: str (Строка)

    Пример: Robot_1() >> Robot_2()

Состав задач процесса

Блок разворачивает каждого робота в группу задач Airflow: в веб-интерфейсе она отображается как Группа задач с идентификатором из параметра Название / ID робота блока Запустить Робота. Кроме групп роботов процесс содержит две служебные задачи, которые обеспечивают монопольный доступ к виртуальной машине.

Граф процесса в веб-интерфейсе Airflow: служебные задачи слева и группа задач робота со всеми возможными задачами запуска.

image_7

Служебные задачи процесса:

  1. hold_pool - сенсор, который занимает слот пула на всё время работы процесса. Задача остаётся в состоянии выполнения, пока не завершатся все конечные задачи роботов, и только после этого освобождает слот;
  2. wait_slot - сенсор, который пропускает роботов дальше только после того, как слот пула занят текущим запуском. Если слот занят другим процессом, запуск ожидает освобождения не дольше значения параметра Время ожидания пула, с.

Задачи внутри группы робота:

  1. create_rdp_session - открывает RDP-сессию на машине робота. Создаётся, если в блоке Запустить Робота включён параметр Выполнять RDP подключение;
  2. upload_script - создаёт рабочую папку запуска на машине робота и копирует в неё файл или папку робота с сервера оркестратора;
  3. get_session_id - определяет идентификатор пользовательской сессии Windows, в которой будет запущен робот. Создаётся только для Windows-машин, где запуск выполняется через PsExec;
  4. execute_script - запускает робота и ожидает завершения. Задача передаёт роботу параметры запуска и завершается ошибкой, если робот завершился с ошибкой или превысил значение параметра Таймаут (в секундах) выполнения робота;
  5. kill_rdp_session - закрывает RDP-сессию. Создаётся вместе с create_rdp_session и выполняется независимо от результата предыдущих задач;
  6. cleanup_script - удаляет рабочую папку запуска с машины робота. Выполняется и после успешного запуска, и после ошибки;
  7. clean_xcom - очищает служебные данные обмена между задачами (XCom). Создаётся только для Airflow 2: в Airflow 3 задачи отрезаны от метабазы, и задача не генерируется;
  8. status_watcher - отмечает группу задач как завершённую с ошибкой, если хотя бы одна из задач робота завершилась неудачно. При успешном запуске задача пропускается.

Задачи робота выполняются последовательно: create_rdp_session >> upload_script >> get_session_id >> execute_script >> kill_rdp_session >> cleanup_script. Порядок самих групп роботов задаётся параметром Порядок выполнения роботов.

Пример использования

Для использования созданного робота в оркестраторе Airflow, необходимо сохранить процесс (процессы) как Python-скрипт.

Сохранение созданного робота как Python-скрипт Сохранение созданного робота как Python-скрипт. Подробнее...

В данном примере процесс my_process запускает двух роботов последовательно, а после них запускает другой процесс оркестратора:

  1. Имя пула - указано значение my_pool: на время работы роботов процесс занимает слот этого пула, поэтому роботы других процессов на ту же машину не заходят. Ожидание слота ограничено 600 секундами, удержание - 7200 секундами;

  2. В разъём Роботы помещён блок Запустить группу роботов с Названием / ID группы group_1. Группа объединяет роботов, которые выполняются на одной машине и связаны общим порядком запуска;

  3. Внутри группы расположены два блока Запустить робота - Robot_1 и Robot_2. Для каждого указаны:

    • Использовать подключение - SSH-подключение SSH_163 из раздела Подключения оркестратора;
    • Путь к роботу на сервере - путь до Python-скрипта робота в папке scripts сервера Airflow;
    • Название главного процесса - имя файла точки входа робота;
    • Директория загрузки робота - путь до установленной студии Puzzle RPA на машине робота;
    • Тип ОС - Linux, Таймаут (в секундах) выполнения робота - 300;

  4. Порядок выполнения роботов внутри группы - указано значение Robot_1() >> Robot_2(): второй робот стартует только после успешного завершения первого;

  5. Следом за группой расположен блок Запустить робот-триггер с Названием / ID триггера puzzle_trigger. Он запускает процесс my_process_2 и, так как параметр Ожидать завершения выполнения процесса выключен, не ждёт его завершения;

  6. Порядок выполнения роботов процесса - указано значение group_1() >> puzzle_trigger(): сначала выполняется группа роботов, затем запускается процесс-триггер.

image_2
Сохранение созданного процесса (Dag) В данном разделе описывается алгоритм создания, сохранения и загрузки в оркестратор созданного процесса (DAG). Подробнее...

Результат

Дерево задач процесса в веб-интерфейсе Airflow: служебные задачи hold_pool и wait_slot, группа роботов group_1 с группами Robot_1 и Robot_2, а также группа триггера puzzle_trigger.

Роботы примера запускаются на Linux-машине без RDP-подключения, поэтому в группах роботов нет задач create_rdp_session, get_session_id и kill_rdp_session.

image_3