Главная
Arenadata Streaming
Сервисы
NiFi
Интеграция
NiFi ADB Connector
Пример записи данных в ADB

Пример записи данных в ADB с помощью NiFi ADB Connector

Ольга Семме

Содержание

Обзор
Предварительные требования
- ADS
- ADP
- ADB
Подключение к серверу PostgreSQL (ADP)
Подключение к Greengage DB (ADB)
Запуск потока данных

Обзор

Для иллюстрации работы NiFi ADB Connector в статье показана реализация загрузки данных из таблиц базы данных PostgreSQL в таблицы ADB (на основе Greengage DB).

Создание NiFi ADB Connector выполняется в пользовательском интерфейсе NiFi и доступно начиная с ADS 3.9.0.1.b1.

Предварительные требования

Ниже описано окружение, используемое для создания NiFi ADB Connector.

Кластер ADS развернут согласно руководству Online-установка. Минимальная версия ADS — 3.9.1.2.b1.
Сервисы NiFi и ZooKeeper установлены в кластере ADS.

При настройке сервисов DBCPConnectionPool для подключения к серверам PostgreSQL (ADP) и Greengage DB (ADB) используются значения, связанные с конфигурацией кластеров ADS, ADP и ADB:

Database Connection URL — ссылка в формате jdbc:postgresql://<host>:5432/<database>, где:
- <host> — хост, на котором установлен компонент ADPG (для подключения к серверу PostgreSQL) или компонент ADB Master (для подключения к серверу Greengage DB);
- <database> — наименование базы данных, в которой создана используемая таблица.
Пример ссылки: jdbc:postgresql://10.92.38.119:5432/adb.
Database Driver Class Name — имя класса драйвера JDBC PostgreSQL, который позволяет программам подключаться к базе данных PostgreSQL, используя стандартный код Java (например, org.postgresql.Driver).
Database Driver Location(s) — место расположения файла драйвера в формате file:<path><driver_name>, где:
- <path> — путь к двоичному JAR-файлу драйвера, размещенному на хостах, где установлен NiFi.
- <driver_name> — имя JAR-файла драйвера, соответствующего используемой версии Greengage DB или PostgreSQL.
Пример указания пути к файлу драйвера, соответствующего ADB 6.30.0: file:/tmp/postgresql-42.2.27.jar.

Команда, приведенная ниже, позволяет загрузить необходимую версию JAR-файла в нужную директорию:
```
$ wget https://jdbc.postgresql.org/download/postgresql-42.2.27.jar
```

ADP

Кластер ADP развернут согласно руководству Online-установка.
На установленном сервере PostgreSQL (кластер ADP) выполнены подготовительные действия:
- Пользователь с именем my_user, привилегиями SUPERUSER и паролем создан в БД postgres.
- В БД postgres создана таблица my_table и в нее добавлены несколько строк с данными.
- Файл pg_hba.conf настроен для обеспечения доступа пользователя с хоста, на котором установлен сервис NiFi кластера ADS. Для этого в поле PG_HBA на странице конфигурационных параметров сервиса ADPG добавлена запись об адресе хоста и пользователе:
  host postgres my_user 10.92.43.206/32 trust

ПРИМЕЧАНИЕ

Для получения информации о работе в ADP обратитесь к статьям:

ADB

Кластер ADB развернут согласно руководству Online-установка.
Сервис ADB установлен в кластере ADB.
Пользователь с именем new_user, привилегиями SUPERUSER и паролем создан в БД adb.
В БД adb создана таблица my_table для записи данных.
Файл pg_hba.conf настроен для обеспечения доступа пользователя с хоста, на котором установлен сервис NiFi кластера ADS. Для этого в поле PG_HBA на странице конфигурационных параметров сервиса ADB добавлена запись об адресе хоста и пользователе:
```
host    adb  new_user       10.92.40.128/24     trust
```
В Interconnect-сети, к которой подключены хосты кластера, должен быть установлен MTU = 9000*(jumbo frame), чтобы пакеты, формируемые ADB (gp_max_packet_size + overhead), помещались в эти фреймы целиком. Для получения более подробной информации о требованиях к сети кластера ADB обратитесь к статье Требования к сети.

ПРИМЕЧАНИЕ

Для получения информации о работе в ADB обратитесь к статьям:

Подключение к серверу PostgreSQL (ADP)

Создайте процессор QueryDatabaseTable и откройте его конфигурацию. Этот процессор выполняет SQL-запросы для получения данных из таблицы PostgreSQL. Заполните параметры, связанные с используемой таблицей PostgreSQL.

Конфигурация процессора QueryDatabaseTable

Конфигурация процессора QueryDatabaseTable
Перейдите в поле значения параметра Database Connection Pooling Service, во всплывающем списке выберите Create new service… и в открывшемся окне создайте экземпляр сервиса DBCPConnectionPool для создания подключения к PostgreSQL server.

Создание экземпляра сервиса DBCPConnectionPool

Создание экземпляра сервиса DBCPConnectionPool
После сохранения созданного экземпляра нажмите , в открывшемся окне NiFi Flow Configuration → Controller Services откройте конфигурацию сервиса и введите параметры, связанные с базой данных PostgreSQL.

Конфигурация сервиса PostgressDBCPConnectionPool

Конфигурация сервиса PostgressDBCPConnectionPool

Создание экземпляров сервиса может быть выполнено до создания процессора. Это может быть использовано для подключения нескольких процессоров к одному сервису. Для создания сервиса выполните:

Правой кнопкой мыши кликните в пустом поле потока и выберите Configure в открывшемся контекстном меню.
В окне NiFi Flow Configuration перейдите на вкладку Controller Services и кликните на +.
Выберите нужный сервис из списка и создайте экземпляр сервиса в открывшемся окне.

Подключение к Greengage DB (ADB)

Для выполнения подключения к таблице Greengage создайте процессор PutGreengageRecord, откройте его конфигурацию и заполните параметры, связанные с таблицей.

Конфигурация процессора PutGreengageRecord

Конфигурация процессора PutGreengageRecord
Перейдите в поле значения параметра Gpfdist Service, во всплывающем списке выберите Create new service… и в открывшемся окне создайте экземпляр сервиса StandardGpfdistService.

Создание экземпляра сервиса StandardGpfdistService

Создание экземпляра сервиса StandardGpfdistService
После сохранения созданного экземпляра нажмите , в открывшемся окне NiFi Flow Configuration → Controller Services откройте конфигурацию сервиса и заполните необходимые параметры.

Конфигурация сервиса StandardGpfdistService

Конфигурация сервиса StandardGpfdistService
Перейдите в поле значения параметра Database Connection Pooling Service, во всплывающем списке выберите Create new service… и в открывшемся окне создайте экземпляр сервиса DBCPConnectionPool для создания подключения к Greengage DB.
После сохранения созданного экземпляра нажмите , в открывшемся окне NiFi Flow Configuration → Controller Services откройте конфигурацию сервиса и введите параметры, связанные с базой данных ADB.

Конфигурация сервиса GreengageDBCPConnectionPool

Конфигурация сервиса GreengageDBCPConnectionPool
Закройте окно NiFi Flow Configuration → Controller Services и снова перейдите к конфигурации процессора PutGreengageRecord. В поле значения параметра Record Reader создайте экземпляр сервиса AvroReader для чтения записей из PostgreSQL в формате Avro со встроенной схемой.

Если используется несколько процессоров PutGreengageRecord для переноса данных из разных источников в одну базу ADB, для значения Gpfdist Service выберите один и тот же созданный StandardGpfdistService.

После создания все сервисы отображаются на странице NiFi Flow Configuration → Controller Services.

Созданные сервисы

Сервис StandardGpfdistService отображается в статусе Invalid до запуска связанного с ним сервиса GreengageDBCPConnectionPool.

Запуск потока данных

Создайте и настройте подключение между процессорами.

Созданные и соединенные процессоры

Процессоры отображаются с ошибками из-за того, что сервисы, связанные с ними, не запущены.

Для запуска потока:

Поочередно запустите сервисы на странице NiFi Flow Configuration → Controller Services, кликнув на иконку .
Запустите созданный поток данных.

Используя запросы к базе данных ADB, можно прочитать полученные данные.

Нашли ошибку? Выделите текст и нажмите Ctrl+Enter чтобы сообщить о ней