Обработка паттернов в данных

Паттерн данных – это комбинация событий, условий и корреляций между этими событиями для отслеживания различных закономерностей и выявления событий. Поиск паттернов используются для анализа и мониторинга потока событий в реальном времени, что позволяет оперативно реагировать на изменения и принимать важные решения. В системах анализа данных паттерн данных формулирует правило, по которому система определяет, соответствует ли входящий поток событий определенным критериям, приводя к срабатыванию определенных действий или уведомлений.

Разберем практический пример обработки паттернов в потоке данных, создаваемым IoT-устройством с кнопками и событиями, вызываемыми при их активации. Нам нужно найти и обработать следующую последовательность нажатия кнопок: «кнопка 1», «кнопка 2», «кнопка 3». Данные передаются в виде JSON-строк, которые с помощью привязок к данным распределяются по колонкам ts и button потока данных input_stream.

Структура передаваемых данных:

{"ts": 1700000000, "button": 1, "device_id": 1, "zone_id": 24}
{"ts": 1701000000, "button": 2, "device_id": 2, "zone_id": 12}

Тело SQL-запроса к YQL:

SELECT * FROM bindings.input_stream MATCH_RECOGNIZE ( -- выполняем поиск паттернов из потока input_stream
    PARTITION BY device_id, zone_id -- разбиваем данные на группы по колонкам device_id и zone_id
    ORDER BY ts -- просматриваем события в порядке возрастания значения колонки ts
    MEASURES
        LAST(B1.ts) AS b1, -- в результатах запроса будем получать последний момент нажатия на кнопку 1
        LAST(B3.ts) AS b3  -- в результатах запроса будем получать последний момент нажатия на кнопку 3
    ONE ROW PER MATCH            -- будем получать одну строку результатов на найденное совпадение
    AFTER MATCH SKIP TO NEXT ROW -- после обнаружения паттерна перейдем на следующую строку
    PATTERN (B1 B2+ B3)      -- ищем паттерн в данных, состоящий из одного нажатия на кнопку 1, одного или нескольких нажатий на кнопку 2 и одного нажатия на кнопку 3
    DEFINE
        B1 AS B1.button = 1, -- определяем переменную B1 как событие нажатия кнопки 1 (значение поля button равно 1)
        B2 AS B2.button = 2, -- определяем переменную B2 как событие нажатия кнопки 2 (значение поля button равно 2)
        B3 AS B3.button = 3  -- определяем переменную B3 как событие нажатия кнопки 3 (значение поля button равно 3)
);

СинтаксисСинтаксис

Команда MATCH_RECOGNIZE выполняет поиск данных по заданному паттерну и возвращает найденные результаты. SQL-синтаксис команды MATCH_RECOGNIZE:

MATCH_RECOGNIZE (
    [ PARTITION BY <партиция_1> [ ... , <партиция_N> ] ]
    [ ORDER BY ключ_сортировки_1 [ ... , ключ_сортировки_N ] ]
    [ MEASURES <выражение_1> AS <имя_колонки_1> [ ... , <выражение_N> AS <имя_колонки_N> ] ]
    [ ONE ROW PER MATCH | ALL ROWS PER MATCH ]
    [ AFTER MATCH SKIP (TO NEXT ROW | PAST LAST ROW) ]
    PATTERN (<шаблон_для_поиска>)
    DEFINE <переменная_1> AS <предикат_1> [ ... , <переменная_N> AS <предикат_N> ]
)

Описание элементов SQL-синтаксиса команды MATCH_RECOGNIZE:

• DEFINE – условия, которым должны соответствовать строки для каждой из переменных <переменная_1> AS <предикат_1> [ ... , <переменная_N> AS <предикат_N> ].
• PATTERN – шаблон для поиска данных. Состоит из переменных и правил поиска паттерна, описанного в выражении <шаблон_для_поиска>. PATTERN работает схожим образом с регулярными выражениями.
• MEASURES определяет список выходных столбцов. Каждый столбец из списка <выражение_1> AS <имя_колонки_1> [ ... , <выражение_N> AS <имя_колонки_N> ] – отдельная конструкция, задающая выходные колонки и описывающая выражения для их вычисления.
• ROWS PER MATCH определяет объем выходных данных по каждому найденному совпадению.
• AFTER MATCH SKIP определяет способ перехода к месту поиска следующего совпадения.
• ORDER BY определяет сортировку входных данных. Поиск паттернов выполняется внутри данных, упорядоченных в соответствии со списком колонок или выражениями, перечисленными в ключ_сортировки_1 [ ... , ключ_сортировки_N ].
• PARTITION BY разделяет входной поток данных по заданным правилам в соответствии с <партиция_1> [ ... , <партиция_N> ]. В каждой из частей поиск паттернов производится независимо.

DEFINEDEFINE

DEFINE <переменная_1> AS <предикат_1> [ ... , <переменная_N> AS <предикат_N> ]

С помощью DEFINE объявляются переменные, которые ищутся во входных данных. Переменные – это имена SQL-выражений, вычисляющихся поверх входных данных. По смыслу SQL-выражения в DEFINE совпадают с поисковыми выражениями SQL-конструкции WHERE. Например, выражение button = 1 выполняет поиск всех строк, содержащих колонку button со значением 1. В качестве условий могут выступать любые SQL-выражения, с помощью которых можно выполнять поиск, включая функции агрегации (LAST, FIRST). Например, button > 2 AND zone_id < 12 или LAST(button) > 10.

В SQL-выражениях обязательно нужно указывать название переменной, для которой производится поиск совпадений. Например в следующей SQL-команде, для условия button = 1 необходимо указать название переменной, для которой производится вычисление - A.

DEFINE
    A AS A.button = 1

При обработке очередной строки данных производится вычисление всех логических выражений всех переменных ключевого слова DEFINE. Если при вычислении выражений переменных DEFINE окажется, что логическое выражение принимает значение ИСТИНА (TRUE), то такая строка получает метку с названием переменной DEFINE и добавляется к списку рассматриваемых на совпадение с паттернами.

ПримерПример

При описании переменных в SQL-выражениях можно использовать ссылки на другие переменные:

DEFINE
    A AS A.button = 1 AND LAST(A.zone_id) = 12,
    B AS B.button = 2 AND FIRST(A.zone_id) = 12

Строка входных данных будет вычислена как переменная A, если в ней присутствует колонка button со значением 1, а в последней строке из множества ранее совпавших с переменной A есть колонка zone_id со значением 12. Строка будет вычислена как переменная B, если в строке данных присутствует колонка button со значением 2, а в первой строке из множества ранее совпавших с переменной A есть колонка zone_id со значением 12.

PATTERNPATTERN

PATTERN (<шаблон_для_поиска>)

Ключевое слово PATTERN описывает поисковый паттерн данных в виде, вычисляющийся на основе переменных из блока DEFINE. Синтаксис PATTERN схож с синтаксисом регулярных выражений.

В PATTERN можно использовать квантификаторы. Они в регулярных выражениях определяют число повторений элемента или подпоследовательности в паттерне для нахождения совпадения. Мы будем использовать переменные A, B, C и D из блока DEFINE для описания квантификаторов. Приведем перечень поддерживаемых квантификаторов:

Поддерживаемые последовательности поиска паттернов:

ПримерПример

PATTERN (B1 E* B2+ B3)
DEFINE
    B1 as B1.button = 1,
    B2 as B2.button = 2,
    B3 as B3.button = 3

В блоке DEFINE описаны переменные B1, B2, B3. Переменная E в блоке DEFINE не описана. Такая запись позволяет интерпретировать E как любое событие, поэтому будет искаться следующий паттерн: одно нажатие кнопки 1, одно или более нажатий кнопки 2 и одно нажатие кнопки 3. При этом между нажатием кнопки 1 и кнопки 2 может быть произвольное число любых других событий.

MEASURESMEASURES

MEASURES <выражение_1> AS <имя_колонки_1> [ ... , <выражение_N> AS <имя_колонки_N> ]

MEASURES описывает набор возвращаемых колонок при нахождении паттерна. Набор возвращаемых колонок должен быть представлен SQL-выражением c агрегирующими функциями над переменными, объявленными в конструкции DEFINE.

ПримерПример

Входными данными для примера являются:

{"ts": 100, "button": 1, "device_id": 3, "zone_id": 0}
{"ts": 200, "button": 1, "device_id": 3, "zone_id": 1}
{"ts": 300, "button": 2, "device_id": 2, "zone_id": 0}
{"ts": 400, "button": 3, "device_id": 1, "zone_id": 1}

MEASURES
    AGGREGATE_LIST(B1.zone_id * 10 + B1.device_id) AS ids,
    COUNT(DISTINCT B1.zone_id) AS count_zones,
    LAST(B3.ts) - FIRST(B1.ts) AS time_diff,
    42 AS meaning_of_life
PATTERN (B1+ B2 B3)
DEFINE
    B1 AS B1.button = 1,
    B2 AS B2.button = 2,
    B3 AS B3.button = 3

Результат:

Колонка ids содержит список значений zone_id * 10 + device_id, посчитанных среди совпавших с переменной B1 строк. Колонка count_zones содержит количество уникальных значений колонки zone_id среди совпавших с переменной B1 строк. Колонка time_diff содержит разницу между значением колонки ts в последней строке из множества совпавших с переменной B3 и значением колонки ts в первой строке из множества совпавших с переменной B1. Колонка meaning_of_life содержит константу 42. Таким образом, выражение в MEASURES может содержать агрегирующие функции над несколькими переменными, но внутри одной агрегирующей функции должна быть только одна переменная.

ROWS PER MATCHROWS PER MATCH

ROWS PER MATCH определяет количество строк результата для каждого найденного паттерна. Режим по умолчанию — ONE ROW PER MATCH.

ONE ROW PER MATCH устанавливает режим работы ROWS PER MATCH на вывод одной строки на найденный паттерн. Схемой данных результата будет являться объединение колонок партиционирования и измерений.

ALL ROWS PER MATCH устанавливает режим работы ROWS PER MATCH на вывод всех строк найденного паттерна, кроме явно исключенных скобками в паттерне. Схемой данных результата будет являться объединение исходных колонок и колонок измерений.

ПримерыПримеры

Входными данными для всех примеров являются:

{"button": 1, "ts": 100}
{"button": 2, "ts": 200}
{"button": 3, "ts": 300}

Пример 1Пример 1

MEASURES
    FIRST(B1.ts) AS first_ts,
    FIRST(B2.ts) AS mid_ts,
    LAST(B3.ts) AS last_ts
ONE ROW PER MATCH
PATTERN (B1 {- B2 -} B3)
DEFINE
    B1 AS B1.button = 1,
    B2 AS B2.button = 2,
    B3 AS B3.button = 3

Результат:

Пример 2Пример 2

MEASURES
    FIRST(B1.ts) AS first_ts,
    FIRST(B2.ts) AS mid_ts,
    LAST(B3.ts) AS last_ts
ONE ROW PER MATCH
PATTERN (B1 {- B2 -} B3)
DEFINE
    B1 AS B1.button = 1,
    B2 AS B2.button = 2,
    B3 AS B3.button = 3

Результат:

AFTER MATCH SKIPAFTER MATCH SKIP

AFTER MATCH SKIP определяет способ перехода от найденного совпадения к поиску следующего совпадения. Поддерживаются только режимы AFTER MATCH SKIP TO NEXT ROW и AFTER MATCH SKIP PAST LAST ROW. Режим по умолчанию — PAST LAST ROW.

ПримерыПримеры

Входными данными для всех примеров являются:

{"button": 1, "ts": 100}
{"button": 1, "ts": 200}
{"button": 2, "ts": 300}
{"button": 3, "ts": 400}

Пример 1Пример 1

MEASURES
    FIRST(B1.ts) AS first_ts,
    LAST(B3.ts) AS last_ts
AFTER MATCH SKIP TO NEXT ROW
PATTERN (B1+ B2 B3)
DEFINE
    B1 AS B1.button = 1,
    B2 AS B2.button = 2,
    B3 AS B3.button = 3

Результат:

Пример 2Пример 2

MEASURES
    FIRST(B1.ts) AS first_ts,
    LAST(B3.ts) AS last_ts
AFTER MATCH SKIP PAST LAST ROW
PATTERN (B1+ B2 B3)
DEFINE
    B1 AS B1.button = 1,
    B2 AS B2.button = 2,
    B3 AS B3.button = 3

Результат:

ORDER BYORDER BY

ORDER BY ключ_сортировки_1 [ ... , ключ_сортировки_N ]

ключ_сортировки ::= { <имена_колонок> | <выражение> }

ORDER BY определяет сортировку входных данных. То есть перед выполнением всех операций поиска паттернов, данные будут предварительно отсортированы по указанным ключам или выражениям. Синтаксис аналогичен SQL-выражению ORDER BY.

ПримерПример

ORDER BY CAST(ts AS Timestamp)

Особенности сортировки в потоковых запросахОсобенности сортировки в потоковых запросах

Потоковые данные потенциально бесконечны, поэтому для их сортировки используется упорядочивание на окне (TimeOrderRecover). Для этого алгоритма в качестве столбцов сортировки можно задавать ровно одно выражение с типом Timestamp, причем сортировка допустима только в направлении возрастания значений, ASC. Управлять размерами окна можно с помощью следующих параметров:

• TimeOrderRecoverAhead, размерность микросекунды, default = -10'000'000 (-10s), должно быть < 0;
• TimeOrderRecoverDelay, размерность микросекунды, default = 10'000'000 (10s), должно быть > 0;
• TimeOrderRecoverRowLimit, размерность число строк, default = 1'000'000, должно быть > 0.

Сортировка на окне работает по следующему алгоритму:

• Если событие не попало в окно, то оно проходит дальше без сортировки.
• Если событие попало в левую половину окна, то оно попадает в сортировку.
• Если число событий в окне превышает TimeOrderRecoverRowLimit, то самое старое событие окна покидает TimeOrderRecover и попадает в MATCH_RECOGNIZE.
• Если событие попало в правую половину окна, то окно сдвигается вправо, и новоприбывшее событие становится Latest row. Все события, выпавшие из окна, покидают TimeOrderRecover и попадают в MATCH_RECOGNIZE.

ПримерыПримеры

Пример 1Пример 1

Не важно, что события могут сильно убегать вперед, но при этом важно, что могут быть отставания в час:

PRAGMA config.flags("TimeOrderRecoverDelay", "-3600000000"); -- -1 час
PRAGMA config.flags("TimeOrderRecoverAhead", "3155760000000000"); -- 100 лет

Пример 2Пример 2

Не хочется, чтобы события из далекого будущего ломали окно сортировки, при этом известно, что события гарантированно приходят хотя бы раз в час:

PRAGMA config.flags("TimeOrderRecoverDelay", "-3600000000"); -- -1 час
PRAGMA config.flags("TimeOrderRecoverAhead", "3600000000"); -- 1 час

PARTITION BYPARTITION BY

PARTITION BY <партиция_1> [ ... , <партиция_N> ]

партиция ::= { <имена_колонок> | <выражение> }

PARTITION BY разбивает входные данные по списку полей, указанных в этом ключевом слове. Выражение превращает исходные данные в несколько независимых групп потоков, в каждом из которых независимо производится поиск паттернов. Если выражение не указывается, то все данные обрабатываются в виде единой группы.

ПримерПример

PARTITION BY device_id, zone_id

ОграниченияОграничения

Степень поддержки команды MATCH_RECOGNIZE со временем будет соответствовать стандарту SQL-2016, однако сейчас действует следующий набор ограничений:

• ORDER_BY. В потоковых запросах в качестве столбцов сортировки можно задавать ровно одно выражение с типом Timestamp, причем сортировка допустима только в направлении возрастания значений, ASC.
• MEASURES. не поддерживаются функции PREV/NEXT.
• AFTER MATCH SKIP. Поддерживаются только режимы AFTER MATCH SKIP TO NEXT ROW и AFTER MATCH SKIP PAST LAST ROW.
• PATTERN. Не реализованы Union pattern variables.
• DEFINE. Не поддерживаются агрегатные функции.