Обзор NiFi ADB Connector

Ольга Семме

Обзор

NiFi ADB Connector обеспечивает высокоскоростную параллельную запись данных между NiFi и Arenadata DB (ADB).

В пользовательском интерфейсе NiFi для создания NiFi ADB Connector доступны объекты разработки Arenadata:

Архитектура NiFi ADB Connector

Ниже на рисунке показано взаимодействие объектов NiFi ADB Connector между собой.

Архитектура NiFi ADB Connector
Архитектура NiFi ADB Connector
Архитектура NiFi ADB Connector
Архитектура NiFi ADB Connector

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

StandardGpfdistService подключается к ADB через сервис пула подключений к базе данных GreengageDBCPConnectionPool.

Gpfdist — утилита Greengage DB для чтения и записи данных из файлов, расположенных на удаленных серверах. Она установлена на всех хостах кластера ADB и обеспечивает параллельную загрузку данных, распределяя их между сегментами равномерно или согласно заданному ключу распределения данных.

Описание конфигурационных параметров сервиса StandardGpfdistService приведено ниже.

Для управления данными к сервису StandardGpfdistService могут быть подключены процессоры:

  • PutGreengageRecord — записывает данные в Greengage DB.

    Для обеспечения контроля над загрузкой сегментов Greengage DB и распределением потоков нагрузки между узлами NiFi используется модель параллелизма.

    Процессоры создают параллельные потоки данных и передают их на сегменты Greengage DB. Обработка записей выполняется асинхронно относительно HTTP-потока gpfdist.

    Описание конфигурационных параметров процессора PutGreengageRecord приведено ниже.

  • GetGreengageRecord —  загружает данные из Greengage DB.

    Процессоры на каждом узле NiFi создают определенное количество задач, которые выполняются параллельно и выгружают фрагменты данных из сегментов Greengage DB. Количество потоков процессора, которые будут обрабатывать эти задачи, можно настроить с помощью параметра Concurrent Tasks на вкладке SCHEDULING.

    Процессоры поддерживают режим пошаговой выгрузки, при котором каждый цикл процессора читает только вновь появившиеся диапазоны данных, вместо того чтобы перечитывать всю таблицу. Такой подход снижает нагрузку на Greengage DB и NiFi, уменьшает объем передаваемых данных и делает повторное опрашивание предсказуемым для сценариев реального времени.

    Описание конфигурационных параметров процессора GetGreengageRecord приведено ниже.

PutGreengageRecord

Ниже приведены параметры процессора PutGreengageRecord.

Параметр Обязательность Описание Значение по умолчанию

Record Reader

true

Ссылка на запуск одного из сервисов контроллера: CSVReader, AvroReader или другого, в зависимости от источника входных данных

 — 

Gpfdist Service

true

Ссылка на запущенный сервис контроллера StandardGpfdistService

 — 

Schema Name

false

Имя схемы, куда будут загружены данные

null

Table Name

true

Имя таблицы, в которую будут загружены данные

 — 

Table Columns

true

Столбцы таблицы, куда будут загружены данные

 — 

Greengage Segment Concurrency Multiplier

false

Целочисленный множитель для параллелизма сегментов Greengage DB, количество потоков загрузки на сегмент. Возможные значения:

  • 1 — рекомендуемое значение;

  • 2 — более высокая параллельность, большее давление на сегменты и gpfdist.

1

Maximum Record Processor Threads

false

Максимальное количество потоков, используемых для обработки записей. Этот параметр должен быть установлен с учетом значений:

  • количество сегментов Greengage DB;

  • значение параметра Greengage Segment Concurrency Multiplier;

  • значение Concurrent Tasks на вкладке SCHEDULING.

8

Flow File Batching Enabled

false

При установке параметра в значение true процессор будет группировать несколько FlowFile в одну операцию INSERT на основе размера пакета

false

Flow Files Batch Size

false

Максимальный общий размер (в байтах) объектов FlowFile, включаемых в одну загрузку. Мягкое ограничение: последний FlowFile может его превышать

10 MB

GetGreengageRecord

Ниже приведены параметры процессора GetGreengageRecord.

Параметр Обязательность Описание Значение по умолчанию

Gpfdist Service

true

Ссылка на запущенный сервис контроллера StandardGpfdistService

 — 

Schema Name

false

Имя схемы, куда будут загружены данные

null

Table Name

true

Имя таблицы, в которую будут загружены данные

 — 

Table Columns

true

Столбцы таблицы, в которые будут загружены данные

 — 

Record Writer

true

RecordSetWriterFactory, которая будет использоваться для записи записей

 — 

Node Parallel Factor

false

Количество параллельных задач/внешних таблиц для выгрузки данных на узел

1

Read Batch Record Count

false

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

10000

Record Buffer Size Per Task

false

Размер буфера обрабатываемых разобранных записей CSV на задачу

10000

Pull Buffer Records Timeout

false

Максимальный таймаут (в миллисекундах) для ожидания появления хотя бы одной записи в очереди буфера при генерации объектов FlowFile. Чем меньше таймаут, тем быстрее текущий поток NiFi освободит задачу и ресурсы. Не рекомендуется устанавливать это значение более 100-500 мс, чтобы не замедлять получение записей из очередей буфера, так как другие задачи могут уже иметь готовые данные

100

Max FlowFiles Per Trigger

false

Максимальное количество объектов FlowFile, создаваемых за один вызов метода выгрузки параллельным потоком NiFi. Это ограничение помогает экономить ресурсы процессора и уменьшать частые переключения контекста между потоками NiFi, выполняющими задачи выгрузки

10

Maximum-value Columns Names

false

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

Требования к указанным в списке столбцам:

  • столбец должен существовать в исходной таблице;

  • столбец не может содержать NULL;

  • тип столбца должен поддерживаться для отслеживания максимального значения (smallint, integer, bigint, real, double precision, numeric, date, time, timestamp, timestamp with time zone).

 — 

StandardGpfdistService

Ниже приведены параметры сервиса контроллера StandardGpfdistService.

Параметр Обязательность Описание Значение по умолчанию

Listening Port

true

Порт для прослушивания входящих запросов gpfdist

80

Database Connection Pooling Service

true

Ссылка на настроенный сервис контроллера DBCPConnectionPool

 — 

Maximum Gpfdist Request Processor Threads

false

Максимальное количество потоков, используемых для обработки запроса gpfdist

8

Minimum Gpfdist Server Threads

false

Минимальное количество потоков, используемых для запуска сервера gpfdist

1

Maximum Gpfdist Server Threads

false

Максимальное количество потоков, используемых для запуска сервера gpfdist

8

Maximum Gpfdist Server Threads Idle Timeout

false

Максимальное время простоя потоков сервера gpfdist в миллисекундах

60000

Total Nifi Nodes

true

Общее количество узлов кластера NiFi

1

Gpfdist Per Greengage Segment Stream Max Buffer Size

false

Максимальный объем данных, который может быть буферизован в памяти для одного потока записи gpfdist на сегмент Greengage DB до применения backpressure. Это ограничивает количество данных "in-flight" (в процессе выполнения) при записи данных в Greengage DB через gpfdist

32 MB

Gpfdist Per Greengage Segment Stream Enqueue Timeout

false

Максимальное время ожидания при попытке добавить данные в буфер потока записи gpfdist для сегмента Greengage DB. Если время ожидания превышено, операция завершается с ошибкой, чтобы предотвратить бесконечное блокирование при backpressure

200 ms

GreengageDBCPConnectionPool

GreengageDBCPConnectionPool — сервис для подключения к ADB. Представляет собой сервис пула подключений к базе данных.

Поддерживаемые типы данных

Сопоставление типов данных ADB с типами полей записей NiFi приведено ниже.

Тип данных ADB Тип данных поля записи NiFi Комментарий

BIT

BOOLEAN

 — 

BOOLEAN

BOOLEAN

 — 

SMALLINT

SHORT, INT

Размер значения должен быть меньше 2 байт

INTEGER

INT

 — 

BIGINT

BIGINT, LONG

 — 

REAL

FLOAT

 — 

DOUBLE

DOUBLE

 — 

NUMERIC(p, s)

DECIMAL

 — 

CHAR(n)

STRING

 — 

VARCHAR(n)

STRING

 — 

ENUM

STRING

 — 

BYTEA

BYTE[], STRING

Для типа STRING значение должно быть записано в шестнадцатеричном формате, например: \xd078

DATE

DATE

 — 

TIME(n)

TIME

 — 

TIMESTAMP(n)

TIMESTAMP

 — 

TIMESTAMPTZ(n)

TIMESTAMP

 — 

MONEY

DECIMAL, STRING, DOUBLE, FLOAT

 — 

UUID

STRING

 — 

JSONB

STRING

 — 

HSTORE

STRING, MAP(STRING, STRING)

Для типа STRING значение должно быть в формате key=value, key1=value1

ARRAY

STRING, STRING[]

Для типа STRING значение должно быть в формате {value, value1}
ПРИМЕЧАНИЕ

Настройка и использование NiFi ADB Connector в ADS при помощи интерфейса NiFi описаны в статье Пример записи данных в ADB.
Нашли ошибку? Выделите текст и нажмите Ctrl+Enter чтобы сообщить о ней