Рабочий процесс
Рабочий процесс — событийно-ориентированное приложение, состоящее из интеграционных и управляющих шагов и переходов между ними. Возможные атрибуты шага:
- название и описание;
- фильтры для входных и выходных данных — jq-выражения, фильтрующие состояние рабочего процесса, которое передается в шаг или является результатом его работы;
- набор действий, которые нужно сделать во время выполнения шага;
- таймаут выполнения;
- политика повторных попыток, которая применяется, если шаг завершился с ошибкой. Если шаг не может быть выполнен из-за исчерпания квоты или его выполнение завершилось с ошибкой, Workflows будет пытаться повторно выполнить шаг согласно политике повторных попыток. В итоге шаг может выполниться успешно или завершиться с ошибкой, если последняя попытка запуска окажется неуспешной. При завершении шага с ошибкой весь запуск переходит в статус
Ошибкаи все дальнейшие шаги не выполняются.
Шаги описываются в спецификации YaWL. В описании каждого шага, кроме последнего, должен быть указан следующий шаг.
В настройках рабочего процесса можно указать:
- пользовательскую сеть, в которой он будет выполняться. Рабочий процесс при этом будет иметь доступ к ресурсам в этой сети.
- сервисный аккаунт, который будет использоваться для получения доступа к закрытым ресурсам, например потокам данных Yandex Data Streams.
Рабочий процесс можно запустить с помощью консоли управления, CLI, API или EventRouter. Подробнее см. Запуск рабочего процесса.
Состояние рабочего процесса
Начальное состояние рабочего процесса — это JSON-объект с ключом input, значение которого — входные данные рабочего процесса. Если входные данные являются JSON-объектом, дополнительно к ключу input добавляются все ключи из входных данных рабочего процесса.
Примеры входных данных и начальных состояний рабочего процесса
| Входные данные | Состояние рабочего процесса |
|---|---|
{"a": "b", "c": 12} |
{"input": {"a": "b", "c": 12}, "a": "b", "c": 12} |
[1, 2, 3] |
{"input": [1, 2, 3]} |
Во время выполнения запуска:
- состояние рабочего процесса передается в качестве входных данных (
input) в шаг; - выходные данные (
output) шагов добавляются в состояние рабочего процесса в том порядке, в котором выполняются шаги.
Используя jq-выражения, можно фильтровать:
- состояние рабочего процесса, которое передается в качестве входных данных (
input) в шаг; - выходные данные (
output), которые добавляются в состояние рабочего процесса.
Примечание
Состояние рабочего процесса в каждый момент времени является JSON-объектом.
Выходные данные каждого шага добавляются в состояние рабочего процесса путем слияния их с текущим состоянием по верхнеуровневым ключам. Поэтому возможна перезапись части данных состояния. Например, если перед запуском шага состояние было JSON-объектом вида {“numbers“: [1,2,3,4], “strings”: [“a”, “b”, “c”]}, а выходные данные шага — {“strings”: [“d”, “e”]}, новым состоянием станет JSON-объект вида {“numbers“: [1,2,3,4], “strings”: [“d”, “e”]}.
Результат выполнения рабочего процесса — выходные данные последнего шага, для которого возможно задать значение поля output.
Состояние рабочего процесса при выполнении шага Parallel
Для шага Parallel состояние рабочего процесса копируется в каждую ветку и в рамках каждой ветки изменяется независимо. Выходными данными шага Parallel до фильтрации является объект, где ключи — уникальные имена веток выполнения, а значения — выходные данные последних шагов соответствующих веток.
Состояние рабочего процесса при выполнении шага Foreach
После фильтрации входных данных шаг Foreach принимает на вход JSON-массив, состоящий из JSON-объектов. После выполнения последовательности шагов в do для каждого элемента массива формируется результат. Результат представляет собой JSON-объект, который складывается в массив на ту же позицию, где находился исходный JSON-объект. Это значит, что для шага Foreach поля input и output обязательны:
input— для преобразования состояния из JSON-объекта в массив;output— для преобразования результата выполнения шага из массива в JSON-объект, который можно добавить в состояние рабочего процесса.
Примеры использования
- Автоматическая загрузка данных в Yandex SpeechSense с помощью Yandex Workflows
- Настройка рабочего процесса Workflows с интеграцией с Yandex Tracker, Yandex Foundation Models и Yandex Cloud Postbox