Потоковая агрегация табличных данных с Flink SQL на Ververica

Как с помощью Flink SQL организовать потоковую агрегацию данных из таблицы PostgreSQL: знакомство с API таблиц в Ververica Cloud на практическом примере.

API таблиц Ververica Cloud: создаем внешние источники и приемники данных

Как я недавно рассказывала, немецкая фирма Ververica создала высокопроизводительный облачный сервис для обработки данных в реальном времени на базе Apache Flink. Этот полностью управляемый облачный сервис позволяет запускать приложения Apache Flink, в т.ч. на бесплатном тарифе. Чтобы познакомиться с возможностями этой платформы поближе, сегодня рассмотрим пример потоковой агрегации данных в одной из таблиц PostgreSQL. Поскольку источником данных является реляционная СУБД, логично из всех API фреймворка выбрать именно табличный, т.е. Flink SQL. Он позволяет подключиться к внешним системам для чтения и записи как пакетных, так и потоковых таблиц. Таблица-источник обеспечивает доступ к данным, которые хранятся во внешних системах (таких как база данных, хранилище значений ключей, очередь сообщений или файловая система). Таблица-приемник передает таблицу во внешнюю систему хранения. В зависимости от типа источника и приемника API таблиц фреймворка может работать с различными форматами: JSON, CSV, AVRO, Parquet, ORC и пр.

Прежде всего, необходимо создать схему источника данных с помощью source-коннектора. В Ververica Cloud это делается в разделе Catalogs.

Создание таблицы Flink SQL на основе коннектора

Затем следует задать схему таблицы и учетные данные подключения. Я буду работать с таблицей jwts, где сохраняются JWT-токены, генерируемые для авторизации менеджеров интернет-магазина. О проектировании и реализации этой системы я писала в блоге нашей Школы прикладного бизнес-анализа здесь и здесь. Токен имеет ограниченный срок действия, и записывается в cookie-файл заголовка HTTP-запроса, который клиент отправляет на сервер. Сам сохраняется в таблице jwts базы данных PostgreSQL, которая развернута в облачной платформе Neon.

Потоковая обработка данных с помощью Apache Flink

Код курса

FLINK

Ближайшая дата курса

1 августа, 2024

Продолжительность

16 ак.часов

Стоимость обучения

48 000 руб.

Выполняем запросы Flink SQL

Запрос Flink SQL на создание таблицы-источника данных выглядит так:

CREATE TABLE `system`.`default`.`jwts` (
  `id` INT,
  `published` VARCHAR(2147483647),
  `token` VARCHAR(2147483647),
  `sysuser` INT
)
COMMENT 'jwts'
WITH (
  'connector' = 'postgres-cdc',
  'database-name' = 'neondb',
  'debezium.plugin.name' = 'pgoutput',
  'hostname' = 'my-host',
  'password' = 'my-pass',
  'port' = '5432',
  'schema-name' = 'public',
  'slot.name' = 'flink',
  'table-name' = 'jwts',
  'username' = 'my-user'
)

После его выполнения будет создана таблица источника данных.

Метаданные и схема созданной таблицы Flink SQL

Чтобы посмотреть ее содержимое средствами Flink SQL, надо задать следующий запрос уже в разделе SQL Editor:

CREATE TABLE print_table( 
  `id` INT,
  `published` STRING,
  `token` STRING,
  `sysuser` INT
) WITH ( 
'connector' = 'print', 
'logger' = 'true' 
); 
INSERT INTO print_table 
SELECT * FROM jwts;

Для выполнения потоковых агрегаций над этой таблицей необходимо создать новый запрос. Чтобы вычислить, сколько раз каждый пользователь входил в систему необходимо сделать группировку по полю sysuser. Для этого запрос будет следующим:

CREATE TABLE print_table( 
 `sysuser` INT,
 `q` BIGINT NOT NULL
) WITH ( 
'connector' = 'print', 
'logger' = 'true' 
); 
INSERT INTO print_table 
SELECT sysuser, COUNT(*) FROM jwts
GROUP BY sysuser;

Таким образом, пользователь с id 1003 входил в систему чаще всего.

Чтобы поэкспериментировать с sink-коннектором, я решила записать в NoSQL-хранилище типа key-value Redis JWT-токены, выдаваемые пользователям. Ключом будет id пользователя, т.е. поле sysuser, а значением – сам токен. Поскольку Redis позволяет задать разные типы данных для значений, лучше всего для этого случая подходит хэш-массив, где к одному ключу можно записать разные поля. Хэши Redis — это типы записей, структурированные как коллекции пар поле-значение. Для этого создала следующий запрос Flink SQL:

CREATE TABLE redis_sink (
      rkey INT,
      rvalue STRING
    ) with (
      'connector' = 'redis',
      'mode' = 'hashmap',
      'host' = 'my-host',
      'port' = 'my-port',
      'password' = 'my-password'
    );

   INSERT INTO redis_sink
    SELECT sysuser, token
    FROM jwts;

Результат выполнения запроса отображается в интерфейсе Flink SQL.

Таблица Flink SQL с sink-коннектором Redis

Однако, в самом хранилище этих данных не оказалось. Как это исправить, расскажу в следующий раз. В заключение отмечу, что выполнение каждой задачи и задания Flink со всеми системными метриками можно посмотреть в GUI фреймворка.