Первый commit

AGENTS.md

@@ -0,0 +1,49 @@
+# Repository Guidelines
+
+This repo is a small Docker Compose environment for learning Apache NiFi with NiFi Registry, Postgres, Kafka, and Kafka UI.
+
+## Project Structure & Module Organization
+
+- `docker-compose.yml`: single entrypoint defining all services and ports.
+- `drivers/`: JDBC drivers mounted into the NiFi container (e.g. `drivers/postgresql-42.7.4.jar`).
+- `nifi-templates/`: NiFi templates/flows and helper SQL (`*.json`, `*.xml`, `*.sql`) mounted into Postgres at `/nifi-templates`.
+- `README.md`: usage notes, URLs, and connection strings.
+- (local) `shared-folder/`: host folder mounted to `/opt/nifi/nifi-current/ls-target` for file-based demos.
+
+## Build, Test, and Development Commands
+
+- `docker compose up`: start the full stack.
+- `docker compose stop`: stop containers without removing volumes.
+- `docker compose start`: start previously stopped containers (preserves all state).
+- `docker compose down`: remove containers and networks.
+- `docker compose down -v`: remove containers, networks, and all named volumes (destructive clean reset).
+- `docker compose logs -f nifi`: follow NiFi logs while debugging processors.
+- `docker compose exec postgres bash -c "export PGPASSWORD=postgres; psql -U postgres -d app"`: open `psql` in the Postgres container.
+
+Notes on state persistence:
+- `postgres`, `nifi`, and `registry` persist state via named volumes in `docker-compose.yml` (survive `docker compose down`, wiped by `docker compose down -v`).
+- Kafka message/topic persistence is disabled by default (see the commented `kafka-data` volume in `docker-compose.yml`).
+
+## Coding Style & Naming Conventions
+
+- YAML: 2-space indentation; keep service names stable (`nifi`, `registry`, `postgres`, `kafka`, `kafka-ui`) to avoid breaking docs.
+- Templates: name files by purpose and target, e.g. `SampleKafka2Postgres.json` and matching `SampleKafka2Postgres.sql`.
+- Docs: keep `README.md` examples runnable and aligned with `docker-compose.yml` ports/URLs.
+
+## Testing Guidelines
+
+No automated tests are currently included. Validate changes by:
+- running `docker compose up` and checking UIs load (NiFi `http://localhost:18443/nifi/`, Registry `http://localhost:18080/nifi-registry`, Kafka UI `http://localhost:8082/`);
+- importing/updating a template from `nifi-templates/` and performing a small end-to-end smoke run.
+If you need a clean slate for validation, use `docker compose down -v` (this wipes Postgres/NiFi/Registry state).
+
+## Commit & Pull Request Guidelines
+
+- Commit messages in history are short, descriptive, and written in Russian (e.g. “Доработки документации”, “Добавлены Postgres, Kafka…”). Follow the same style.
+- PRs: include a short description, the motivation (issue link if applicable), and note any port/credential changes. Add screenshots for UI-facing doc updates when helpful.
+
+## Security & Configuration Tips
+
+This is for local learning, not production. Defaults include:
+- NiFi single-user creds in `docker-compose.yml` (`admin` / `Password123456`);
+- Postgres creds (`postgres` / `postgres`) and published port `5437`.

README.md

@@ -0,0 +1,147 @@
+# nifi-docker
+
+Учебный стенд для знакомства с Apache NiFi и Kafka: NiFi + NiFi Registry + Postgres + Kafka + Kafka UI.
+
+## Быстрый старт
+```sh
+docker compose up -d
+docker compose ps
+```
+
+Остановить стенд:
+```sh
+docker compose stop
+```
+
+Запустить обратно (после `stop` состояние сохраняется):
+```sh
+docker compose start
+```
+
+Удалить контейнеры и сеть:
+```sh
+docker compose down
+```
+Затем поднять обратно (с сохранённым состоянием в томах):
+```sh
+docker compose up -d
+```
+
+Удалить ещё и тома (volumes) — полный сброс (деструктивно):
+```sh
+docker compose down -v
+```
+
+### Что сохраняется между перезапусками
+- `docker compose stop/start` — сохраняется всё (контейнеры не удаляются).
+- `docker compose down` — контейнеры удаляются, но тома (volumes) остаются: сохраняются NiFi (`conf`/`state`), NiFi Registry, Postgres.
+- `docker compose down -v` — полный сброс: удаляются и контейнеры, и тома (volumes).
+
+Примечание: в этой конфигурации Kafka-сообщения/топики не сохраняются между `docker compose down` → `up` (чтобы не копить дисковое пространство). Между `stop` → `start` Kafka сохраняется. Пример подключения тома (volume) для Kafka есть в `docker-compose.yml`.
+
+## Адреса и доступы
+- NiFi: http://localhost:18443/nifi/ (логин `admin`, пароль `Password123456`)
+- NiFi docs: https://nifi.apache.org/documentation/
+- Registry: http://localhost:18080/nifi-registry
+- Registry docs: https://nifi.apache.org/docs/nifi-registry-docs/
+- Kafka UI: http://localhost:8082/
+- Kafka UI docs: https://docs.kafka-ui.provectus.io/
+
+## Важно про адреса (внутри Docker и с хоста)
+Если вы настраиваете подключение *в NiFi*, то `localhost` почти всегда будет неправильным (NiFi живёт в контейнере).
+
+Используйте имена сервисов из `docker-compose.yml`:
+- Postgres (из NiFi): `jdbc:postgresql://postgres:5432/app`
+- Kafka (из NiFi): `kafka:29092`
+- Registry (из NiFi): `http://registry:18080`
+
+А с локальной машины:
+- Postgres: `jdbc:postgresql://localhost:5437/app`
+- Kafka: `localhost:9092`
+
+## PostgreSQL
+JDBC-драйвер для Postgres лежит в `drivers/` и монтируется в контейнер NiFi как `/opt/nifi/nifi-current/drivers/`.
+
+Параметры для DBCP в NiFi:
+- Database Connection URL: `jdbc:postgresql://postgres:5432/app`
+- Database Driver Class Name: `org.postgresql.Driver`
+- Database Driver Location(s): `/opt/nifi/nifi-current/drivers/postgresql-42.7.4.jar`
+- Database User: `postgres`
+- Password: `postgres`
+
+### Подключение через DBeaver (удобнее всего)
+Postgres проброшен наружу на порт `5437`, поэтому в DBeaver создайте подключение со следующими параметрами:
+- Host: `localhost`
+- Port: `5437`
+- Database: `app`
+- Username: `postgres`
+- Password: `postgres`
+
+Если DBeaver предложит скачать драйвер — соглашайтесь скачать/установить драйвер PostgreSQL.
+
+Инициализация демо-схем/таблиц:
+```sh
+docker compose exec -T postgres psql -U postgres -d app -f /nifi-templates/SampleKafka2Postgres.sql
+```
+Запускайте это после первого старта или после `docker compose down -v` (скрипт не идемпотентный: при повторном запуске будут ошибки про существующие схемы/таблицы).
+
+### Через консоль (если нужно)
+Просмотр данных:
+```sh
+docker compose exec -it postgres bash -c "export PGPASSWORD=postgres; psql -U postgres -d app"
+select * from ods.samplekafka2postgres order by id desc limit 10;
+```
+
+## Kafka
+- С локальной машины (например, для консольных утилит): `localhost:9092`
+- Из NiFi (внутри Docker): `kafka:29092`
+
+## Примеры flow (шаблоны)
+В `nifi-templates/` лежат примеры:
+- `Sample2Kafka.xml` / `Sample2Kafka.json` — публикует сообщения в Kafka topic `Sample2Kafka`
+- `SampleKafka2Postgres.json` — читает из Kafka topic `Sample2Kafka` и пишет в Postgres (в `stg.samplekafka2postgres`, затем вызывает `ods.load_samplekafka2postgres()`)
+
+### Памятка: как импортировать Process Group / Flow в NiFi
+Файлы нужно загружать через браузер с вашей машины (каталог `nifi-templates/`).
+
+**Вариант 1: шаблон `.xml` (Template)**
+1) Откройте NiFi: http://localhost:18443/nifi/
+2) В верхнем меню найдите `Templates` → `Upload Template` → выберите файл `nifi-templates/Sample2Kafka.xml`.
+3) На канвасе: нажмите правой кнопкой мыши → `Instantiate Template` (или иконка Template на панели) → выберите шаблон → разместите process group на канвасе.
+
+**Вариант 2: flow definition `.json`**
+Название пункта может отличаться в зависимости от UI/версии, но смысл один — “загрузить process group/flow definition из файла”:
+1) В верхнем меню найдите действие вроде `Upload` / `Import` / `Process Group` → выберите загрузку из файла.
+2) Выберите `nifi-templates/SampleKafka2Postgres.json` (или `Sample2Kafka.json`) и разместите process group на канвасе.
+
+После импорта, как правило, нужно:
+- перейти внутрь process group;
+- включить Controller Services (Configure → `Controller Services` → Enable, или “enable all controller services”);
+- затем запустить процессоры.
+
+Рекомендуемый минимальный сценарий:
+1) topic `Sample2Kafka` вручную создавать обычно не требуется: он создаётся автоматически при первой попытке записи (если в Kafka включено автосоздание топиков; по умолчанию оно включено). Если по какой-то причине topic не появился — создайте его в Kafka UI.
+2) Импортируйте flow в NiFi (в зависимости от UI: import/upload template для `.xml` или import flow definition для `.json`). Если вы импортировали раньше, то после `docker compose down` flow сохранится.
+3) Внутри flow включите Controller Services, затем запустите процессоры.
+
+Kafka UI уже настроен в `docker-compose.yml`:
+- Cluster name: `Kafka Cluster`
+- Bootstrap Servers: `kafka:29092`
+
+## Shared folder (общая папка)
+Каталог `shared-folder/` на хосте смонтирован в контейнер NiFi как `/opt/nifi/nifi-current/ls-target` (удобно для ListFile/GetFile).
+Если после запуска контейнеров возникают проблемы с правами: `sudo chown -R $USER shared-folder`.
+
+## Полезные команды
+```sh
+docker compose logs -f nifi
+docker compose logs -f kafka
+docker compose exec postgres bash
+```
+
+Для доступа из NiFi к сервисам на локальной машине используйте `host.docker.internal` вместо `localhost`.
+
+## Если что-то не работает
+- NiFi может запускаться 1–3 минуты; смотрите `docker compose logs -f nifi`.
+- Если процессор в NiFi не подключается к Kafka/Postgres, проверьте, что используете адреса “из NiFi” (см. раздел про Docker).
+- Если порты заняты, измените проброс портов в `docker-compose.yml`.

docker-compose.yml

@@ -0,0 +1,96 @@
+services:
+
+  nifi:
+    image: apache/nifi:1.27.0
+    ports:
+      - 18443:8443 # Unsecured HTTP Web Port
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    environment:
+      - SINGLE_USER_CREDENTIALS_USERNAME=admin
+      - SINGLE_USER_CREDENTIALS_PASSWORD=Password123456
+      - NIFI_WEB_HTTP_PORT=8443
+    volumes:
+      - ./drivers:/opt/nifi/nifi-current/drivers
+      - ./shared-folder:/opt/nifi/nifi-current/ls-target
+      - nifi-conf:/opt/nifi/nifi-current/conf
+      - nifi-state:/opt/nifi/nifi-current/state
+      # Если хочешь сохранять очереди/контент/провенанс (обычно для обучения не нужно) — раскомментируй:
+      # - nifi-flowfile-repo:/opt/nifi/nifi-current/flowfile_repository
+      # - nifi-content-repo:/opt/nifi/nifi-current/content_repository
+      # - nifi-provenance-repo:/opt/nifi/nifi-current/provenance_repository
+      # - nifi-database-repo:/opt/nifi/nifi-current/database_repository
+
+  registry:
+    image: apache/nifi-registry:1.27.0
+    ports:
+      - 18080:18080
+    environment:
+      - NIFI_REGISTRY_WEB_HTTP_PORT=18080
+    volumes:
+      - registry-db:/opt/nifi-registry/nifi-registry-current/database
+      - registry-flow-storage:/opt/nifi-registry/nifi-registry-current/flow_storage
+      
+  postgres:
+    image: postgres:16.4
+    # restart: always
+    ports:
+      - 5437:5432
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: app
+    volumes:
+      - pgdata:/var/lib/postgresql/data
+      - ./nifi-templates:/nifi-templates
+    
+  kafka:
+    image: apache/kafka:3.8.0
+    hostname: kafka
+    container_name: kafka
+    ports:
+      - 9092:9092
+    # Если хочешь сохранять топики/сообщения между `docker compose down/up` — раскомментируй:
+    # volumes:
+    #   - kafka-data:/tmp/kraft-combined-logs
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_NODE_ID: 1
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093
+      KAFKA_LISTENERS: PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,PLAINTEXT_HOST://0.0.0.0:9092
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
+      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
+
+  kafka-ui:
+    container_name: kafka-ui
+    image: provectuslabs/kafka-ui:v0.7.2
+    ports:
+      - 8082:8080
+    depends_on:
+      - kafka
+    environment:
+      DYNAMIC_CONFIG_ENABLED: "true"
+      KAFKA_CLUSTERS_0_NAME: "Kafka Cluster"
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: "kafka:29092"
+      KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL: "PLAINTEXT"
+
+volumes:
+  pgdata:
+  nifi-conf:
+  nifi-state:
+  registry-db:
+  registry-flow-storage:
+  # nifi-flowfile-repo:
+  # nifi-content-repo:
+  # nifi-provenance-repo:
+  # nifi-database-repo:
+  # kafka-data:

nifi-templates/Sample2Kafka.json

@@ -0,0 +1 @@
+{"flowContents":{"identifier":"2c55a545-02c0-34b0-9cf9-d702a87a935a","instanceIdentifier":"095ddaa1-0192-1000-a2e3-f8c334e01a9c","name":"Sample2Kafka","comments":"","position":{"x":1040.0,"y":80.0},"processGroups":[],"remoteProcessGroups":[],"processors":[{"identifier":"2246dfec-0602-3e50-9af1-456dbfd1dba2","instanceIdentifier":"6acf9814-30c2-35d7-8cf5-6af3d4da8a76","name":"PublishKafkaRecord_2_6","comments":"","position":{"x":320.0,"y":216.0},"type":"org.apache.nifi.processors.kafka.pubsub.PublishKafkaRecord_2_6","bundle":{"group":"org.apache.nifi","artifact":"nifi-kafka-2-6-nar","version":"1.27.0"},"properties":{"compression.type":"none","attribute-name-regex":null,"record-writer":"8655a15b-8c8a-3432-9d3d-3f824cdea257","bootstrap.servers":"kafka:29092","sasl.kerberos.principal":null,"sasl.kerberos.service.name":null,"kerberos-credentials-service":null,"partition":null,"record-key-writer":null,"sasl.mechanism":"GSSAPI","max.block.ms":"5 sec","record-reader":"12b2d414-0b35-39c7-bf7c-68c9abbc7d10","message-header-encoding":"UTF-8","kerberos-user-service":null,"transactional-id-prefix":null,"ack.wait.time":"5 secs","sasl.username":null,"use-transactions":"false","acks":"1","aws.profile.name":null,"Record Metadata Strategy":"Use Configured Values","security.protocol":"PLAINTEXT","ssl.context.service":null,"max.request.size":"1 MB","sasl.token.auth":"false","message-key-field":null,"Failure Strategy":"Route to Failure","partitioner.class":"org.apache.kafka.clients.producer.internals.DefaultPartitioner","sasl.kerberos.keytab":null,"topic":"Sample2Kafka","publish-strategy":"USE_VALUE"},"propertyDescriptors":{"compression.type":{"name":"compression.type","displayName":"Compression Type","identifiesControllerService":false,"sensitive":false},"attribute-name-regex":{"name":"attribute-name-regex","displayName":"Attributes to Send as Headers (Regex)","identifiesControllerService":false,"sensitive":false},"record-writer":{"name":"record-writer","displayName":"Record Writer","identifiesControllerService":true,"sensitive":false},"bootstrap.servers":{"name":"bootstrap.servers","displayName":"Kafka Brokers","identifiesControllerService":false,"sensitive":false},"sasl.kerberos.principal":{"name":"sasl.kerberos.principal","displayName":"Kerberos Principal","identifiesControllerService":false,"sensitive":false},"sasl.kerberos.service.name":{"name":"sasl.kerberos.service.name","displayName":"Kerberos Service Name","identifiesControllerService":false,"sensitive":false},"kerberos-credentials-service":{"name":"kerberos-credentials-service","displayName":"Kerberos Credentials Service","identifiesControllerService":true,"sensitive":false},"partition":{"name":"partition","displayName":"Partition","identifiesControllerService":false,"sensitive":false},"record-key-writer":{"name":"record-key-writer","displayName":"Record Key Writer","identifiesControllerService":true,"sensitive":false},"sasl.mechanism":{"name":"sasl.mechanism","displayName":"SASL Mechanism","identifiesControllerService":false,"sensitive":false},"max.block.ms":{"name":"max.block.ms","displayName":"Max Metadata Wait Time","identifiesControllerService":false,"sensitive":false},"record-reader":{"name":"record-reader","displayName":"Record Reader","identifiesControllerService":true,"sensitive":false},"message-header-encoding":{"name":"message-header-encoding","displayName":"Message Header Encoding","identifiesControllerService":false,"sensitive":false},"kerberos-user-service":{"name":"kerberos-user-service","displayName":"Kerberos User Service","identifiesControllerService":true,"sensitive":false},"transactional-id-prefix":{"name":"transactional-id-prefix","displayName":"Transactional Id Prefix","identifiesControllerService":false,"sensitive":false},"ack.wait.time":{"name":"ack.wait.time","displayName":"Acknowledgment Wait Time","identifiesControllerService":false,"sensitive":false},"sasl.username":{"name":"sasl.username","displayName":"Username","identifiesControllerService":false,"sensitive":false},"use-transactions":{"name":"use-transactions","displayName":"Use Transactions","identifiesControllerService":false,"sensitive":false},"acks":{"name":"acks","displayName":"Delivery Guarantee","identifiesControllerService":false,"sensitive":false},"aws.profile.name":{"name":"aws.profile.name","displayName":"AWS Profile Name","identifiesControllerService":false,"sensitive":false},"Record Metadata Strategy":{"name":"Record Metadata Strategy","displayName":"Record Metadata Strategy","identifiesControllerService":false,"sensitive":false},"security.protocol":{"name":"security.protocol","displayName":"Security Protocol","identifiesControllerService":false,"sensitive":false},"ssl.context.service":{"name":"ssl.context.service","displayName":"SSL Context Service","identifiesControllerService":true,"sensitive":false},"max.request.size":{"name":"max.request.size","displayName":"Max Request Size","identifiesControllerService":false,"sensitive":false},"sasl.token.auth":{"name":"sasl.token.aLine truncated

nifi-templates/SampleKafka2Postgres.json

@@ -0,0 +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":"f6cbce4f-3175-33f8-8f23-8abdb7d977e9","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

nifi-templates/SampleKafka2Postgres.sql

@@ -0,0 +1,42 @@
+create schema stg;
+
+CREATE TABLE stg.samplekafka2postgres (
+	dttm timestamptz NULL,
+	txt text NULL
+);
+
+
+create schema ods;
+
+CREATE TABLE ods.samplekafka2postgres (
+	id serial4 NOT NULL,
+	ins_dttm timestamptz default now() NOT NULL,
+	dttm timestamptz NULL,
+	txt text NULL,
+	CONSTRAINT samplekafka2postgres_pkey PRIMARY KEY (id)
+);
+
+create or replace procedure ods.load_samplekafka2postgres()
+language plpgsql 
+as $$
+begin 
+	-- перенос содержимого таблицы в ods полностью
+	insert into ods.samplekafka2postgres(dttm, txt)
+	select dttm, txt
+	from stg.samplekafka2postgres;
+
+	-- удаление делаем в процессе NiFi
+	-- delete from stg.samplekafka2postgres;
+
+	commit;
+end; $$;
+
+-- Примеры запросов
+select * from ods.samplekafka2postgres;
+
+delete from stg.samplekafka2postgres;
+delete from ods.samplekafka2postgres where ins_dttm < now() - interval '10 minute'; 
+
+select * from stg.samplekafka2postgres;
+select * from ods.samplekafka2postgres order by id desc;
+select count(*) from ods.samplekafka2postgres;