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

Основные принципы построения процесса загрузки:

• Поток NIFI максимально универсален для источников

• Метаданные потоков загрузки расположены вне NIFI

• С добавлением нового потока загрузки должен справится аналитик/разработчик БД

В общем виде взаимодействие сиситем можно представить в следующем виде.

На рисунке:

• Источники - все, что поставляет данные. В моей текущей реализации это только базы данных.

• Кеш - либо внутренний кэш NIFI, либо Redis.

• БД метаданых - хранилище инормации о потоках.

• Назначение - приемники данных. У меня сейчас реализованы две БД, и одно файловое хранилище.

Метаданные

Прежде, чем показать поток обработки на NIFI, необходимо уделить внимание метаданным. Метаданные можно хранить любым удобным для вас способом. Например, в виде XML файла в GIT репозитории. В текущей реализации я использую плоскую таблицу в БД.

Структура таблицы (основные элементы потока):

Поток зарузки

Итак, поток NIFI состоит из следующих процесных групп:

Генерация (Shedulers)

В блоке собраны все возможные генераторы, триггеры.

Для удобства я разделил генераторы на группы. Общий смысл данной процесной группы - полуить файл с заполненым атрибутом ${schedule}, по которому будут выбраны потоки из БД. В интервальной группе находятся триггеры, котрые срабатывают по интервалу, например, триггер раз в пять минут генерирует файл с расписанием 5min.

В группе TimeOfDay я генерирую файл раз в минуту, и выставляю ${schedule} = ${now():format("HH:mm")}.

Таким образом, можно поместить проццессор GenerateFlowFile с нужным интервалом или crone выражением, и поток будет загружен только при срабатывании нужного зрачения.

Инициализация (Init)

В данной группе выполняется инициализация потока. Ядром и основным элементом является получение метаданых потока:

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

Запрос простой:

SELECT * from ${paramtable} 
WHERE
active = 1 AND 
'${schedule}' in (
    SELECT value 
    FROM string_split(schedule, ';')
)
ORDER BY priority

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

Также при наличии флага tgt_truncate выполняется удаление целевой таблицы. В атрибут ${database.name} помещается значение из ${src.erp}, это позволит использовать один сервис для проксирования подключенных источников.

Извлекаем данные (Extract)

В этой группе происходит гегерация запросов и выполнение их к источникам. Здесь применяется замечательный компонент DBCPConnectionPoolLookup, позволяющий перечислить все доступные системы, и выполнять доступ к ним по атрибуту ${database.name}.

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

Однако данный процессор работает медленно, по раней мере к нагруженным БД гереация запросы выполнялась за несклько секунд. При повышении требований по скорости обработки решено было сделать свой генератор для основных систем - MSSql и Oracle. Использовал следующее выражение:

SELECT ${column.list:isEmpty():ifElse('*', ${column.list})} 
FROM ${src.schema}.${src.table}
${src.additionalwhere:isEmpty():ifElse(${src.additionalwhere},
                                       ${src.additionalwhere:prepend('WHERE ')})}

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

Если дубликатов нет, то переходим к запросам к источнику. Если задана схема, то запрос идет через ExecuteSQLRecord, в котором Writer сконфигурирован на схему в атрибуте.

Параметры ExecuteSql:

В параметрах я указываю:

1. Max Rows Per Flow File - eсли надо сохранять файл для Clickhouse (так грузим словари), то выгрузку не разбивать, в противном случае задаю значение из глобальной переменной.

2. Output Batch Size - если задан запрос, который надо выполнить в конце загрузки, то выполнять батчевую загрузку, иначе отправлять на выход по пять файлов. Батчевая загрузка позволит отследить количество фрагментов в процессе запроса.

3. Fetch Size - некоторые драйверы умеют принимать этот параметр: количество записей, передаваемых в пакете.

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

Трансформация (Trasnform)

В данный момент у нас применяется одна трансформация - это формирование полей Starge слоя. Эти поля и уже описывал в первой и второй части, покажу лишь небольшую автоматизацию.

Сначала извлекается авто-схема в атрибут ${avro.schema}, для этого есть процессор ExtractAvroMetadata. Далее формируется правило конкатенации полей:

${avro.schema:jsonPath("$.fields..name"):replace('[',''):
replace(']',''):replace('"',''):replace(',',',/'):replace('\r',''):
replace('\n',''):replace(' ',''):prepend('/')}

И с помощью процессора UpdateRecord вычисляется хеш записи, предварительно схема модифицируется:

Загрузка (Load)

Для загрузки применяется процессор PutDataвaseRecord, предварительно в атрибут ${database.name} помещается значение из ${tgt.erp}. Изначально применялся один процессор, но в последствии в целях оптимизации производительности и разделения реалтаймовских потоков от интервальных, предварительно определяется система назначения.

Также есть группа, где потоки сохраняются в виде файлов. В ней обновляются атрибуты ${filename}, конвертируется формат в Parquet и выполняется сохранение в смонтированную NFS директорию. Далее на сервере Clickhouse эти файлы подтягиваются как словари.

Постзагрузка (Postload)

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

Сначала производится вычисление максимума. Для этого применяется процессор QueryRecord, запрос имеет вид:

SELECT MAX(${src.incrementkey}) as MaxValue
FROM FLOWFILE

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

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

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

Дальнейшие шаги являются служебными - формирования лога, выполнение уведомлений в канал Slack и так далее.

Заключение

Я постарался описать сам подход к построению параметризованного универсального потока загрузки. Конечно, у такого подхода есть и плюсы и минусы, и не всегда его можно применить. Но если у вас единый стандарт заполнения DWH, то большинство операций можно свести к одному потоку с небольшим набором ветвлений.

Преимущества

• Все потоки описаны единым образом, для добавления нового потока нет необходимости привлекать разработчика Nifi, весь процесс заключается в добавлении или редактировании записи в таблице.

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

• Поток загрузки может быть запущен с различным расписанием, а не только по интервалу или крону. Также, добавив обработку HTTP запросов и принимая на входе JSON с параметрами, можно гибко формировать условия отбора. Мы таким образом формируем запросы к целевой системе в зависимости от преобразований в Clickhouse, формирование запроса выполняется в Airflow.

• Деплой выполняется обновлением одной верхней процессором группы.

Недостатки

• В данный момент есть одна плоская таблица для хранения метаданных. Ведутся разработки над более гибкой структурой - иерархическим JSON, который позволит подключать настраиваемые трансформации и выходы.

• Требуется тщательно планировать большие исторические загрузки, и не отдавать этот момент аналитикам и разработчикам БД. Однако, если с ресурсами нет проблем, то этот пункт исключается.

• Выполнение потока зависит от доступности внешних систем - кеша, БД метаданных и так далее.

• Для управления метаданными желательно разработать интерфейс. Первые 200 потоков можно редактировать в БД, но потом таблица разрастается, возрастает вероятность ошибки.

Спасибо за то, что прочитали этот текст. Жду ваших комментариев. На вопросы могу ответить в группе телеграмм. Если есть опечатки/ошибки в тексте - сообщайте, устраню.