Настройка SQL адаптера

Обзор

SQL адаптер Accelera Flows - это модуль для получения данных из реляционной БД, в котором каждая строка таблицы принимается за событие.

Поддерживаемые БД:

Postgres
Oracle
MSSQL
MySQL
MariaDB
SQLite3
Amazon Redshift

Алгоритм работы

SQL-адаптер с заданной частотой отправляет запросы в БД, обрабатывая ответ.

Адаптер может получать события в 2-х режимах:

Полного вычитывания данных по запросу.
Получения данных по максимальному значению.

При полном вычитывании данных все строки, которые получит адаптер будут обработаны в соответствии с заданным шаблоном и отправлены как события в Accelera Flows.

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

Конфигурация

Распакуйте образы приложения из файла flows_sql_adapter_{your_version}.tar.gz следующей командой:

docker load < flows_sql_adapter_{your_version}.tar.gz

Создайте файл docker-compose.yaml со следующим содержимым:

version: '3'
services:
  sql_adapter:
    image: devcubesolutions/flows_sql_adapter:latest
    command: npm run start
    container_name: sql_adapter
    restart: unless-stopped
    environment:
      - NODE_ENV=production
      - LOG_LEVEL=info
      - REDIS_CONNECTION=redis://localhost:6379
			- BROKER_CONNECTION=amqp://user:password@broker_host:5672
      - ADAPTER_NAME=sql-transactions-adapter
#     DB connection settings
      - DRIVER=Oracle
      - DB_USER=admin
      - DB_PASSWORD=admin
      - CONNECTION_STRING=localhost:49161/XE
#     SQL Settings
      - STATEMENT=SELECT t.*, ROWID FROM XDB."transactions" t
      - CRON=0 1 * * * *
      - MAX_VALUE=false
      - INCREMENT_COLUMN=PROCESSED_DTTM
      - INCREMENT_TYPE=datetime
      - DATETIME_FORMAT=YYYY-MM-DD HH:mm:ss
      - QUERY_TEMPLATE_PARAM_NAME=maxvaluefield
      - INITIAL_VALUE=2020-01-01 00:00:00
#     Event parsing settings
      - EVENTS_PER_SECOND=300
      - ID_FIELD=CUSTOMER_NO
      - EVENT_NAME=
      - EVENT_FIELD=ACCOUNT_NUMBER
      - FLOW_ID=
      - FLOW_ID_FIELD=ACCOUNT_NUMBER
      - EVENT_CONTEXT=trx_date,{{TRANSACTION_DATE}},amount,{{ORIGINAL_AMOUNT}},acc_number,{{ACCOUNT_NUMBER}},merch_city,{{MERCHANT_CITY}},country,{{MERCHANT_COUNTRY}},mcc,{{MERCHANT_CATEGORY}},datetime,{{PROCESSED_DTTM}}

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

image

Имя образа, который был распакован на предыдущем шаге, можно узнать командой docker images

container_name

Имя контейнера при запуске, рекомендуется оставить без изменения

restart

Политика рестарта процесса при остановке или ошибке, по умолчанию 'unless-stopped'

network_mode

Подробнее читайте в Особенности подключения к Redis.

Environment

Любой из этих параметров (например, DB_PASSWORD) может быть указан в командной строке при запуске адаптера.

REDIS_CONNECTION - адрес подключения к БД Redis. Подробнее читайте в Особенности подключения к Redis.

LOG_LEVEL - уровень логирования (trace, debug, info, warn, error, fatal). По умолчанию - info.

BROKER_CONNECTION - адрес подключения RabbitMQ.

ADAPTER_NAME - имя адаптера. Не должно содержать пробелов или символов (кроме - или _). Должно быть уникально для каждого адаптера.

DRIVER - драйвер подключения к БД. Возможные значения:

Postgres,
Oracle,
MSSQL,
MySQL,
MariaDB,
SQLite3,
AmazonRedshift.

DB_USER - имя пользователя БД.

DB_PASSWORD - пароль к БД.

CONNECTION_STRING - строка подключения к БД.

STATEMENT - строка запроса к БД.

CRON - настройка периодичности запроса в формате cron. Используется шаблон с учетом секунд (* * * * * *).

MAX_VALUE - true - активация режима получения данных по максимальному значению. false - режим полной вычитки. Опционально.

INCREMENT_COLUMN - если MAX_VALUE = true, то необходимо указать название колонки, по которой будет отслеживаться максимальное значение. Колонка должна иметь числовой тип или дата/время. Опционально.

INCREMENT_TYPE - если MAX_VALUE = true, то необходимо указать тип данных параметра INCREMENT_COLUMN. Может быть number или datetime.

DATETIME_FORMAT - если INCREMENT_TYPE = datetime, то необходимо указать формат даты для колонки.

QUERY_TEMPLATE_PARAM_NAME - название параметра, которое будет использоваться в SQL - запросе. На его место подставляется последнее обработанное значение из указанной колонки.

INITIAL_VALUE - начальное значение переменной, QUERY_TEMPLATE_PARAM_NAME.

EVENTS_PER_SECOND - укажите число, если нужно ограничить максимальную пропускную способность адаптера. Оставьте пустым, если такая необходимость отсутствует. Опционально.

ID_FIELD - название поля из таблицы, откуда будет получен id события. Название должно точно совпадать с названием в таблице, включая регистр.

EVENT_NAME - название события, которое будет отправлено в Flow Engine. Применяется для всех событий из таблицы. Если нужно брать название события из таблицы, то оставьте пустым.

EVENT_FIELD - название поля из таблицы, откуда будет получено название события. Название должно точно совпадать с названием в таблице, включая регистр. Если вы указали EVENT_NAME, оставьте это поле пустым.

💡

Если EVENT_NAME или EVENT_FIELD не заданы, то события из таблицы будут игнорироваться.

FLOW_ID - ID сценария, куда будут отправлены события. Если нужно отправить события во все сценарии, оставьте это поле пустым.

FLOW_ID_FIELD - название поля из таблицы, откуда будет получен id сценария. Название должно точно совпадать с названием в таблице, включая регистр. Если вы указали FLOW_ID, оставьте это поле пустым. Если нужно отправить события во все сценарии, оставьте это поле пустым.

EVENT_CONTEXT - укажите шаблон контекста, который будет отправлен в Accelera Flow. Шаблон задается следующим образом: название параметра, который будет в контексте,{{название параметре из таблицы}},...

Например, trx_date,{{TRANSACTION_DATE}} - в контексте события будет присутствовать параметр trx_date. В качестве значения данного параметра будет содержимое колонки TRANSACTION_DATE.

Пары таких параметров укажите через запятую без пробелов. Если параметр в таблице будет отсутствовать или иметь значение null, то в систему Accelera Flow отправится пустое значение.

Запуск

Для запуска файлового адаптера используйте следующую команду:

docker-compose -f {путь к вашему docker-compose файлу} up -d

Получить доступ к логам и проверить работоспособность адаптера можно командой

docker logs -f sql_adapter

Остановка

Для остановки всех адаптеров в yaml-файле используйте команду:

docker-compose -f {путь к вашему docker-compose файлу} down

Для остановки конкретного адаптера используйте команду:

docker-compose -f {путь к вашему docker-compose файлу} stop <имя_адаптера>