Пояснения к потокам NiFi

2026-01-08 13:11:47 +03:00
1 parent fd2fbe5256
commit b111d5136a
@@ -0,0 +1,110 @@
+# Гайд для менти: как работают примеры flow на JSON (NiFi → Kafka → Postgres)
+
+Этот гайд про два flow definition файла:
+- `nifi-templates/Sample2Kafka.json` — генерирует записи и публикует их в Kafka topic `Sample2Kafka` в виде JSON.
+- `nifi-templates/SampleKafka2Postgres.json` — читает JSON из Kafka topic `Sample2Kafka` и пишет данные в Postgres.
+
+## 0) Что мы строим (общая картинка)
+Цепочка очень простая:
+1) NiFi генерирует запись (как “строчку данных”).
+2) NiFi отправляет эту запись в Kafka (как сообщение в topic).
+3) Другой flow в NiFi читает эти сообщения из Kafka.
+4) NiFi фильтрует/готовит данные и вставляет их в Postgres.
+
+Это учебный пример, который показывает базовые роли технологий:
+- **NiFi** — визуальный конвейер (pipeline) данных.
+- **Kafka** — “почтовый ящик”/шина сообщений между системами.
+- **Postgres** — база данных, где данные хранятся и доступны SQL-запросами.
+
+## 1) Мини-словарь NiFi (3 понятия)
+- **FlowFile** — “сообщение”, которое течёт по графу. У FlowFile есть *content* (тело) и *attributes* (метаданные).
+- **Processor** — шаг обработки (прочитать/преобразовать/записать).
+- **Controller Service** — общая настройка, которую используют процессоры (например, “как читать JSON” или “как подключаться к базе”).
+
+## 2) Какой JSON у нас в примере
+Запись состоит из двух полей:
+- `dttm` — дата/время (timestamp).
+- `txt` — строка (text).
+
+В первом flow запись создаётся процессором `GenerateRecord` по схеме (она задана в настройке процессора).
+
+Важно: в этом учебном варианте **мы не используем Avro** и не используем Schema Registry. Сообщения в Kafka — это обычный JSON.
+
+## 3) Перед стартом: что проверить
+1) Стенд поднят: `docker compose up -d`
+2) Открывается NiFi: http://localhost:18443/nifi/ (логин `admin`, пароль `Password123456`)
+3) Открывается Kafka UI: http://localhost:8082/
+4) (Если это первый запуск после `docker compose down -v`) примените SQL-скрипт для демо-таблиц:
+   - `docker compose exec -T postgres psql -U postgres -d app -f /nifi-templates/SampleKafka2Postgres.sql`
+
+## 4) Flow 1 — `Sample2Kafka.json` (NiFi → Kafka)
+Цель: регулярно публиковать новые JSON-сообщения в Kafka topic `Sample2Kafka`.
+
+### Шаги внутри flow
+1) `GenerateRecord`
+   - Создаёт запись по схеме (в настройке `schema-text`).
+   - Частота генерации задаётся расписанием процессора (например, раз в 2 секунды).
+2) `PublishKafkaRecord_2_6`
+   - Берёт записи и публикует в Kafka.
+   - Важные настройки:
+     - `bootstrap.servers`: `kafka:29092` (имя сервиса внутри Docker).
+     - `topic`: `Sample2Kafka`
+     - `record-reader`: `JsonTreeReader` (как “прочитать запись” перед отправкой)
+     - `record-writer`: `JsonRecordSetWriter` (как сериализовать в JSON)
+
+### Как проверить, что flow работает
+- В NiFi у `PublishKafkaRecord_2_6` растёт счётчик “out”.
+- В Kafka UI в topic `Sample2Kafka` появляются сообщения.
+
+Если сообщений нет:
+- Проверьте, что процессоры запущены и нет очереди/ошибок на `failure`.
+- Проверьте, что внутри NiFi указано `kafka:29092`, а не `localhost:9092`.
+
+## 5) Flow 2 — `SampleKafka2Postgres.json` (Kafka → Postgres)
+Цель: читать сообщения из `Sample2Kafka`, отфильтровать записи и записать их в Postgres.
+
+### Шаги внутри flow
+1) `ConsumeKafkaRecord_2_6`
+   - Читает сообщения из Kafka topic `Sample2Kafka`.
+   - Важные настройки:
+     - `bootstrap.servers`: `kafka:29092`
+     - `topic`: `Sample2Kafka`
+     - `record-reader`: `JsonTreeReader` (как разобрать JSON из Kafka)
+2) `QueryRecord`
+   - Выполняет простой “фильтр” по данным.
+   - В примере есть запрос `GOOD_DATE` (это имя relationship), который отбирает записи по условию на `dttm`.
+   - На выходе получаются записи, которые прошли фильтр.
+3) `MergeContent` (micro-batch)
+   - Собирает несколько FlowFile в “пачку” перед загрузкой в БД.
+   - Зачем: много маленьких вставок (каждое сообщение → отдельный insert) может перегружать БД; batching снижает накладные расходы.
+   - Важные настройки:
+     - `Minimum Number of Entries` / `Maximum Number of Entries` — размер пачки (сколько записей объединяем).
+     - `Max Bin Age` — максимальное ожидание, чтобы пачка всё равно отправилась даже если данных мало.
+   - Компромисс: чем больше batch, тем меньше нагрузка на БД, но больше задержка (latency).
+4) `PutDatabaseRecord`
+   - Вставляет записи в таблицу `stg.samplekafka2postgres`.
+   - Использует подключение к Postgres через `DBCPConnectionPool`.
+5) `ExecuteSQL`
+   - Вызывает хранимую процедуру `ods.load_samplekafka2postgres()` и чистит staging.
+   - Это шаг “переложить в витрину/ODS”, чтобы показать типичный паттерн.
+
+### Как проверить результат в Postgres
+В контейнере Postgres выполните:
+- `docker compose exec -it postgres bash -c \"export PGPASSWORD=postgres; psql -U postgres -d app\"`
+
+И затем SQL:
+- `select * from ods.samplekafka2postgres order by id desc limit 10;`
+
+Если таблица пустая:
+- Убедитесь, что в Kafka действительно есть сообщения (проверьте Kafka UI).
+- Посмотрите, нет ли ошибок на `parse.failure` (у Kafka consumer) или `failure` (у DB).
+
+## 6) Типовые проблемы (и куда смотреть)
+- **В NiFi указали `localhost` вместо имени сервиса**: из контейнера `localhost` — это сам контейнер, не Kafka/Postgres.
+- **Controller Services выключены**: внутри process group откройте `Controller Services` и включите нужные сервисы.
+- **Нет таблиц/процедуры**: после “чистого сброса” нужно снова выполнить `SampleKafka2Postgres.sql`.
+
+## 7) Мини-упражнения для закрепления (5–10 минут)
+1) В `Sample2Kafka.json` поменяйте расписание `GenerateRecord` (например, 2 sec → 5 sec) и проверьте скорость появления сообщений.
+2) В `QueryRecord` измените фильтр `GOOD_DATE` (например, более жёсткое/мягкое условие) и проверьте, как меняется количество строк в Postgres.
+3) Добавьте новое поле в схему `GenerateRecord` и протащите его до Postgres (после изменения не забудьте обновить SQL-схему/таблицы).
@@ -97,6 +97,8 @@ select * from ods.samplekafka2postgres order by id desc limit 10;
 - Из NiFi (внутри Docker): `kafka:29092`

 ## Примеры flow (шаблоны)
+Короткий гайд для менти по JSON-версии потоков: `MENTEE_GUIDE_JSON_FLOWS.md`.
+
 В `nifi-templates/` лежат примеры:
 - `Sample2Kafka.json` — публикует JSON-сообщения в Kafka topic `Sample2Kafka`
 - `SampleKafka2Postgres.json` — читает JSON из Kafka topic `Sample2Kafka` и пишет в Postgres (в `stg.samplekafka2postgres`, затем вызывает `ods.load_samplekafka2postgres()`)
@@ -1 +1 @@
-{"flowContents":{"identifier":"c428496e-2726-3000-96bc-6ac1f35698af","instanceIdentifier":"01921000-a73c-1966-17cf-0ddc02bd11b3","name":"SampleKafka2Postgres","comments":"","position":{"x":504.0,"y":336.0},"processGroups":[],"remoteProcessGroups":[],"processors":[{"identifier":"44f50642-4e30-3c0d-993a-f1c5669e0925","instanceIdentifier":"502f3b5a-65ae-1b46-a84e-c2f5fce6f989","name":"ExecuteSQL","comments":"","position":{"x":464.0,"y":816.0},"type":"org.apache.nifi.processors.standard.ExecuteSQL","bundle":{"group":"org.apache.nifi","artifact":"nifi-standard-nar","version":"1.27.0"},"properties":{"esql-max-rows":"0","dbf-default-precision":"10","Max Wait Time":"0 seconds","Database Connection Pooling Service":"3e7c0e44-2b3f-3929-b154-a54ebdb4f672","sql-post-query":"delete from stg.samplekafka2postgres;","esql-auto-commit":"true","dbf-user-logical-types":"false","dbf-default-scale":"0","sql-pre-query":null,"compression-format":"NONE","esql-output-batch-size":"0","esql-fetch-size":"0","SQL select query":"call ods.load_samplekafka2postgres();","dbf-normalize":"false"},"propertyDescriptors":{"esql-max-rows":{"name":"esql-max-rows","displayName":"Max Rows Per Flow File","identifiesControllerService":false,"sensitive":false},"dbf-default-precision":{"name":"dbf-default-precision","displayName":"Default Decimal Precision","identifiesControllerService":false,"sensitive":false},"Max Wait Time":{"name":"Max Wait Time","displayName":"Max Wait Time","identifiesControllerService":false,"sensitive":false},"Database Connection Pooling Service":{"name":"Database Connection Pooling Service","displayName":"Database Connection Pooling Service","identifiesControllerService":true,"sensitive":false},"sql-post-query":{"name":"sql-post-query","displayName":"SQL Post-Query","identifiesControllerService":false,"sensitive":false},"esql-auto-commit":{"name":"esql-auto-commit","displayName":"Set Auto Commit","identifiesControllerService":false,"sensitive":false},"dbf-user-logical-types":{"name":"dbf-user-logical-types","displayName":"Use Avro Logical Types","identifiesControllerService":false,"sensitive":false},"dbf-default-scale":{"name":"dbf-default-scale","displayName":"Default Decimal Scale","identifiesControllerService":false,"sensitive":false},"sql-pre-query":{"name":"sql-pre-query","displayName":"SQL Pre-Query","identifiesControllerService":false,"sensitive":false},"compression-format":{"name":"compression-format","displayName":"Compression Format","identifiesControllerService":false,"sensitive":false},"esql-output-batch-size":{"name":"esql-output-batch-size","displayName":"Output Batch Size","identifiesControllerService":false,"sensitive":false},"esql-fetch-size":{"name":"esql-fetch-size","displayName":"Fetch Size","identifiesControllerService":false,"sensitive":false},"SQL select query":{"name":"SQL select query","displayName":"SQL select query","identifiesControllerService":false,"sensitive":false},"dbf-normalize":{"name":"dbf-normalize","displayName":"Normalize Table/Column Names","identifiesControllerService":false,"sensitive":false}},"style":{},"schedulingPeriod":"0 sec","schedulingStrategy":"TIMER_DRIVEN","executionNode":"ALL","penaltyDuration":"30 sec","yieldDuration":"1 sec","bulletinLevel":"WARN","runDurationMillis":0,"concurrentlySchedulableTaskCount":1,"autoTerminatedRelationships":[],"scheduledState":"ENABLED","retryCount":10,"retriedRelationships":[],"backoffMechanism":"PENALIZE_FLOWFILE","maxBackoffPeriod":"10 mins","componentType":"PROCESSOR","groupIdentifier":"c428496e-2726-3000-96bc-6ac1f35698af"},{"identifier":"02492338-a831-3f03-82aa-d4fff07fbe0c","instanceIdentifier":"628c3237-6526-19d2-ae64-f79cf9bb18d3","name":"ConsumeKafkaRecord_2_6","comments":"","position":{"x":464.0,"y":104.0},"type":"org.apache.nifi.processors.kafka.pubsub.ConsumeKafkaRecord_2_6","bundle":{"group":"org.apache.nifi","artifact":"nifi-kafka-2-6-nar","version":"1.27.0"},"properties":{"header-name-regex":null,"Commit Offsets":"true","output-strategy":"USE_VALUE","record-writer":"5534ae28-0a7a-3e5e-8380-5c1a08507ae9","group.id":"0","bootstrap.servers":"kafka:29092","topic_type":"names","sasl.kerberos.principal":null,"sasl.kerberos.service.name":null,"kerberos-credentials-service":null,"max-uncommit-offset-wait":"1 secs","sasl.mechanism":"GSSAPI","honor-transactions":"true","record-reader":"a87fe6e0-c59b-34e2-8f38-77d7c189a6dd","kerberos-user-service":null,"message-header-encoding":"UTF-8","key-format":"byte-array","sasl.username":null,"max.poll.records":"10000","aws.profile.name":null,"security.protocol":"PLAINTEXT","ssl.context.service":null,"key-record-reader":null,"sasl.token.auth":"false","sasl.kerberos.keytab":null,"Communications Timeout":"60 secs","topic":"Sample2Kafka","separate-by-key":"false","key-attribute-encoding":"utf-8","auto.offset.reset":"latest"},"propertyDescriptors":{"header-name-regex":{"name":"header-name-regex","displayName":"Headers to Add as Attributes (Regex)","identifiesControllerService":false,"sensitive":false},"CLine truncated
+{"flowContents":{"identifier":"c428496e-2726-3000-96bc-6ac1f35698af","instanceIdentifier":"01921000-a73c-1966-17cf-0ddc02bd11b3","name":"SampleKafka2Postgres","comments":"","position":{"x":504.0,"y":336.0},"processGroups":[],"remoteProcessGroups":[],"processors":[{"identifier":"44f50642-4e30-3c0d-993a-f1c5669e0925","instanceIdentifier":"502f3b5a-65ae-1b46-a84e-c2f5fce6f989","name":"ExecuteSQL","comments":"","position":{"x":464.0,"y":816.0},"type":"org.apache.nifi.processors.standard.ExecuteSQL","bundle":{"group":"org.apache.nifi","artifact":"nifi-standard-nar","version":"1.27.0"},"properties":{"esql-max-rows":"0","dbf-default-precision":"10","Max Wait Time":"0 seconds","Database Connection Pooling Service":"3e7c0e44-2b3f-3929-b154-a54ebdb4f672","sql-post-query":"delete from stg.samplekafka2postgres;","esql-auto-commit":"true","dbf-user-logical-types":"false","dbf-default-scale":"0","sql-pre-query":null,"compression-format":"NONE","esql-output-batch-size":"0","esql-fetch-size":"0","SQL select query":"call ods.load_samplekafka2postgres();","dbf-normalize":"false"},"propertyDescriptors":{"esql-max-rows":{"name":"esql-max-rows","displayName":"Max Rows Per Flow File","identifiesControllerService":false,"sensitive":false},"dbf-default-precision":{"name":"dbf-default-precision","displayName":"Default Decimal Precision","identifiesControllerService":false,"sensitive":false},"Max Wait Time":{"name":"Max Wait Time","displayName":"Max Wait Time","identifiesControllerService":false,"sensitive":false},"Database Connection Pooling Service":{"name":"Database Connection Pooling Service","displayName":"Database Connection Pooling Service","identifiesControllerService":true,"sensitive":false},"sql-post-query":{"name":"sql-post-query","displayName":"SQL Post-Query","identifiesControllerService":false,"sensitive":false},"esql-auto-commit":{"name":"esql-auto-commit","displayName":"Set Auto Commit","identifiesControllerService":false,"sensitive":false},"dbf-user-logical-types":{"name":"dbf-user-logical-types","displayName":"Use Avro Logical Types","identifiesControllerService":false,"sensitive":false},"dbf-default-scale":{"name":"dbf-default-scale","displayName":"Default Decimal Scale","identifiesControllerService":false,"sensitive":false},"sql-pre-query":{"name":"sql-pre-query","displayName":"SQL Pre-Query","identifiesControllerService":false,"sensitive":false},"compression-format":{"name":"compression-format","displayName":"Compression Format","identifiesControllerService":false,"sensitive":false},"esql-output-batch-size":{"name":"esql-output-batch-size","displayName":"Output Batch Size","identifiesControllerService":false,"sensitive":false},"esql-fetch-size":{"name":"esql-fetch-size","displayName":"Fetch Size","identifiesControllerService":false,"sensitive":false},"SQL select query":{"name":"SQL select query","displayName":"SQL select query","identifiesControllerService":false,"sensitive":false},"dbf-normalize":{"name":"dbf-normalize","displayName":"Normalize Table/Column Names","identifiesControllerService":false,"sensitive":false}},"style":{},"schedulingPeriod":"0 sec","schedulingStrategy":"TIMER_DRIVEN","executionNode":"ALL","penaltyDuration":"30 sec","yieldDuration":"1 sec","bulletinLevel":"WARN","runDurationMillis":0,"concurrentlySchedulableTaskCount":1,"autoTerminatedRelationships":[],"scheduledState":"ENABLED","retryCount":10,"retriedRelationships":[],"backoffMechanism":"PENALIZE_FLOWFILE","maxBackoffPeriod":"10 mins","componentType":"PROCESSOR","groupIdentifier":"c428496e-2726-3000-96bc-6ac1f35698af"},{"identifier":"02492338-a831-3f03-82aa-d4fff07fbe0c","instanceIdentifier":"628c3237-6526-19d2-ae64-f79cf9bb18d3","name":"ConsumeKafkaRecord_2_6","comments":"","position":{"x":464.0,"y":104.0},"type":"org.apache.nifi.processors.kafka.pubsub.ConsumeKafkaRecord_2_6","bundle":{"group":"org.apache.nifi","artifact":"nifi-kafka-2-6-nar","version":"1.27.0"},"properties":{"header-name-regex":null,"Commit Offsets":"true","output-strategy":"USE_VALUE","record-writer":"5534ae28-0a7a-3e5e-8380-5c1a08507ae9","group.id":"0","bootstrap.servers":"kafka:29092","topic_type":"names","sasl.kerberos.principal":null,"sasl.kerberos.service.name":null,"kerberos-credentials-service":null,"max-uncommit-offset-wait":"1 secs","sasl.mechanism":"GSSAPI","honor-transactions":"true","record-reader":"a87fe6e0-c59b-34e2-8f38-77d7c189a6dd","kerberos-user-service":null,"message-header-encoding":"UTF-8","key-format":"byte-array","sasl.username":null,"max.poll.records":"10000","aws.profile.name":null,"security.protocol":"PLAINTEXT","ssl.context.service":null,"key-record-reader":null,"sasl.token.auth":"false","sasl.kerberos.keytab":null,"Communications Timeout":"60 secs","topic":"Sample2Kafka","separate-by-key":"false","key-attribute-encoding":"utf-8","auto.offset.reset":"latest"},"propertyDescriptors":{"header-name-regex":{"name":"header-name-regex","displayName":"Headers to Add as Attributes (Regex)","identifiesControllerService":false,"sensitive":false},"CLine truncated