Apache NiFi → Elasticsearch и PostgreSQL: поиск и BI-отчёты

Обновлено 2 сентября 2026 года.

Elasticsearch помогает искать по тексту обращений и комментариев, PostgreSQL — хранить связанные сущности и рассчитывать показатели. NiFi может управлять доставкой данных в обе системы. Такой поток нужен, когда требования к поиску действительно выходят за возможности уже используемой базы. Для простого поиска сначала стоит проверить полнотекстовый поиск PostgreSQL: вторая система увеличивает объём сопровождения.

Сначала определите состав данных

Для учебного примера возьмём обращения с полями portal_id, ticket_id, subject, body, status, created_at и updated_at. Это контракт примера, а не готовый ответ любого метода Битрикс24. Ключ обращения — пара portal_id и ticket_id. Если отчёту нужны сумма сделки или ответственный, их необходимо получить и включить в преобразование отдельно.

Определите, какие поля доступны для полнотекстового поиска. Не копируйте в индекс всё содержимое CRM автоматически: права пользователей, персональные данные и срок хранения должны учитываться и в поисковой копии.

Как построить поток NiFi

1. Получить данные. Для REST используйте InvokeHTTP, настройте адрес, авторизацию и обработку ответа. GenerateTableFetch формирует SQL для извлечения частей таблицы из базы; URL метода CRM вместо SQL-источника ему не подходит.
2. Обойти страницы. Сохраняйте параметр продолжения из ответа конкретного API. Один успешный HTTP-запрос не доказывает, что получены все обращения.
3. Проверить записи. Приведите даты и идентификаторы к согласованным типам, отделите ошибки разбора, проверьте обязательный ключ.
4. Сохранить исходный пакет. Архив или надёжная очередь с идентификатором загрузки позволит повторить доставку, если одна из систем недоступна.
5. Обновить обе копии. PostgreSQL получает UPSERT по ключу, Elasticsearch — документ со стабильным _id, например portal_id:ticket_id. Для более старых версий записи предусмотрите защиту от перезаписи свежих данных.

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

Минимальный русский анализатор

Elasticsearch содержит встроенный анализатор russian. Ниже пример тела запроса создания учебного индекса. Выполнять его нужно в тестовом окружении, подставив своё имя индекса:

PUT /crm_tickets_example
{
  "mappings": {
    "properties": {
      "ticket_key": {"type": "keyword"},
      "subject": {"type": "text", "analyzer": "russian"},
      "body": {"type": "text", "analyzer": "russian"},
      "status": {"type": "keyword"},
      "updated_at": {"type": "date"}
    }
  }
}

Имена фильтров russian_stop и russian_stemmer в пользовательском анализаторе требуют собственных определений в settings. Анализатор текста и ingest pipeline — разные сущности: имя анализатора нельзя использовать как имя несуществующего pipeline. Проверьте разбор слов и результаты поиска на реальных формулировках сотрудников.

Что читать из Power BI

Отчёт по числу обращений, статусам и времени обработки удобнее строить на представлениях PostgreSQL. Текстовый поиск можно вынести в отдельный интерфейс. Если найденные документы должны фильтровать BI-отчёт, спроектируйте передачу набора ключей; сам факт наличия двух подключений не связывает их автоматически.

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

Проверка согласованности

Запишите новое обращение, измените его текст и статус, повторите пакет и удалите запись в источнике. Сверьте PostgreSQL и Elasticsearch по одному ключу и версии. Затем отключите одну из систем, восстановите её и повторите недоставленные пакеты. Успешная запись в PostgreSQL не означает успешную индексацию.

Результат приёмки — измеренные задержки, совпадение контрольных наборов и рабочее восстановление. Обещать фиксированное ускорение или время ответа до испытания на ваших данных нельзя.