Быстрый старт в YTsaurus Flow (YQL)

YQL over Flow позволяет описать пайплайн потоковой обработки данных в виде декларативного SQL-запроса — без написания кода на C++, Java, Go или Python. Пайплайн запускается как ванилла операция на выбранном кластере YTsaurus.

Важно

Находится в активной разработке, ещё не вся запланированная функциональность доступна.

Полезные ссылки

Документация YQL — полный справочник по синтаксису YQL
YQL провайдер YT Flow — исходный код

Прагмы

Запрос YQL over Flow управляется через набор прагм, полный список см. в разделе Справочник прагм.

Первый запрос

Пример: построчное преобразование стрима (мап).

-- выбрать движок Flow
PRAGMA Engine = "ytflow";

-- кластер для внутренних таблиц пайплайна
PRAGMA Ytflow.Cluster = "<cluster-name>";
-- кластер для ванилла операции
PRAGMA Ytflow.RuntimeCluster = "<cluster-name>";
-- директория с пайплайнами
PRAGMA Ytflow.PipelineDirectory = "//home/my-project";
-- имя пайплайна
PRAGMA Ytflow.PipelineName = "my-pipeline";
-- число воркеров
PRAGMA Ytflow.WorkerCount = "1";

-- читать из входной очереди, трансформировать, писать в выходную
INSERT INTO
    <cluster-name>.`//home/my-project/sink_queue`
SELECT
    string_field || "_processed" AS string_field,
    int64_field,
    EndsWith(string_field, "bar") AS predicate
FROM
    <cluster-name>.`//home/my-project/source_queue`
WHERE int64_field > 1;

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

Описание возможностей YQL см. в разделе Возможности.

Как запустить

Пререквизиты

Нужно иметь права на чтение и запись во все упоминаемые в запросе директории, а также вычислительную квоту на кластере YTsaurus, указанном как Ytflow.RuntimeCluster.

Есть два способа запустить запрос:

Через UI YTsaurus: откройте вкладку Queries на рантайм кластере и выполните запрос.

Через Python-клиент:

from yt.wrapper import YtClient

# любой продакшн кластер
client = YtClient('<cluster-name>')

# запустить запрос и дождаться завершения
client.run_query(
    engine='yql',
    settings=dict(
        # рантайм кластер передаётся здесь
        cluster='<cluster-name>',
    ),
    query='<YQL query>',
    sync=True,
)

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

Мониторинг

Для отслеживания работы запущенного пайплайна доступны:

Дашборд — вкладка Flow → Monitoring.
Логи контроллера (состояние воркеров, возможные проблемы):
```
 yt --proxy <кластер-пайплайна> flow show-logs //home/my-project/my-pipeline
```
Логи джобов — через ванилла операцию, доступную по ссылке из кубика flowPublish в графе пайплайна.

См. также

URL Downloader

Возможности