Привет, меня зовут Иван и я SDET‑специалист в SimbirSoft. Полагаю, каждый разработчик хотя бы раз слышал про инструмент Kafka, многие работали с ним. Наиболее простое его использование — это настройка взаимодействия между микросервисами. А если попробовать использовать его как инструмент для логирования — связать Kafka и Elasticsearch? И при чём тут вообще Elasticsearch? Всё просто: это система, которая позволяет работать с большим объёмом данных. В нашем случае также необходимо использовать Kibana, надстройку над эластиком, которая позволит визуализировать данные.

В этой статье мы рассмотрим процесс настройки потоковой передачи логов из Kafka в Elasticsearch:

• как настроить взаимодействие между Kafka и Elasticsearch

• как настроить Kibana для визуализации логов, хранящихся в Elasticsearch.

Эта статья будет полезна тестировщикам для настройки анализа логов в распределённых системах, а также разработчикам, желающим реализовать альтернативные сценарии использования Kafka для централизованного логирования.

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

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

Почему просто сохранить логи в файлы недостаточно?

Если логов много, а система распределённая, то:

• логи разбросаны по разным машинам;

• найти нужные данные вручную сложно и долго;

• файлы растут, их становится сложно хранить и обрабатывать.

Почему использовать Elasticsearch удобно:

• всё в одном месте. Логи собираются из разных сервисов и сохраняются централизованно в Elasticsearch;

• быстрый поиск. Elasticsearch позволяет искать по логам за миллисекунды — хоть по тексту сообщения, хоть по времени;

• аналитика и визуализация. В Kibana можно построить графики и дашборды, чтобы видеть «узкие места» в системе;

• надёжность. Если Elasticsearch упадёт, Kafka сохранит логи, и их можно будет отправить позже.

А для чего нужна генерация в реальном времени?

Когда у вас что‑то случится в системе в момент её работы, будет очень сложно прочитать логи и выявить ошибку, если искать файлы с логами вручную. Если будет настроен elastic, то вы сможете увидеть и сразу идентифицировать ошибку.

Что нам нужно для работы:

Apache Kafka:

• установленный Kafka‑брокер и Zookeeper для управления кластером;

• Kafka должен быть настроен для приема данных (топик для логов).

Elasticsearch:

• версия Elasticsearch, совместимая с используемыми клиентами (рекомендуем последнюю стабильную версию);

• базовая конфигурация с поддержкой индексов для логов.

Kibana:

• установленная и настроенная Kibana для визуализации данных из Elasticsearch.

Java и JVM:

• Java Development Kit (JDK 11 или выше) для разработки кастомного Kafka Consumer;

• инструменты сборки проекта, например Maven или Gradle;

• Kafka Client API: Для реализации Kafka Consumer, который будет читать данные из топика;

• Elasticsearch High‑Level REST Client: Для отправки данных в Elasticsearch.

Схема архитектуры

Начнём

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

1. Первым делом скачайте и установите Java. Например, здесь, но можете использовать любой другой источник.

2. Скачайте инструменты: 

• Kafka Scala 2.13 — kafka_2.13-3.9.0.tgz (asc, sha512)

• Скачать Elasticsearch

• Скачать logstash 

• Скачать Kibana Index of Kibana‑local

После создайте несколько папок и назовите их соответственно: kafka, elasticsearch, logstash, kibana. Распакуйте туда ранее скачанные инструменты.

После подготовительной части приступим к настройке каждого инструмента по отдельности:

В бой! Для начала давайте попробуем запустить кафку. 

Перейдите в папку с Kafka и откройте командную строку. (Win + R открыть окно «Выполнить» и ввести CMD):

Для macOS/Linux вы можете воспользоваться BASH и указать путь: 

cd /usr/local/kafka (вставьте верный путь к вашей папке). После этого откройте папку с конфигурациями config:

Далее измените файлы server.properties и zookeeper.properties соответственно следующим образом:

Обратите внимание, вместо tmp нужно вставить путь к вашей папке с Kafka + data. Например, так: dataDir=D:\article\kafka\data\zookeeper

Отлично. Теперь нужно запустить ZooKeeper. Если хотите поглубже узнать что это и для чего используется, приложу ссылку на статью по Kafka: Apache Kafka для чайников / Хабр):

Windows:

\bin\windows\zookeeper‑server‑start.bat.\config\zookeeper.properties

macOS/Linux:

bin/zookeeper‑server‑start.sh config/zookeeper.properties

 

Переходим к самой Kafka: 

Windows: 

\bin\windows\kafka‑server‑start.bat.\config\server.properties

macOS/Linux: 

bin/kafka‑server‑start.sh config/server.properties

Проверьте, всё ли запустилось. Далее нужно создать новый топик. Откройте новое окно и введите команду: 

Windows: 

\bin\windows\kafka‑topics.bat ‑create ‑logs ‑bootstrap‑server localhost:9092 ‑partitions 1 ‑replication‑factor 1

macOS/Linux: 

bin/kafka‑topics.sh ‑create ‑topic logs ‑bootstrap‑server localhost:9092 ‑partitions 1 ‑replication‑factor 1

А так можно посмотреть их список: 

Windows: 

\bin\windows\kafka‑topics.bat ‑list ‑bootstrap‑server localhost:9092

macOS/Linux: 

bin/kafka‑topics.sh ‑list ‑bootstrap‑server localhost:9092

Если у вас всё получилось, давайте попробуем запустить Elastic. На данный момент Kafka будет отложена в сторону, но закрывать её не нужно. Если вы всё же это сделали, то ничего страшного. Далее мы к ней вернёмся и вы снова её запустите. 

Приступим к настройке Elasticsearch. Для этого перейдите в папку, в которую до этого распаковали эту поисковую систему.

После предлагаю для простоты настройки отключить защиту в Elasticsearch. Для этого сделаем следующее: откроем папку с Elasticsearch, а в ней зайдем в папку config, где найдём файл Elasticsearch.yaml:

Откройте его и найдите поля:

xpack.security.enabled:
xpack.security.enrollment.enabled:

По умолчанию значения этих настроек равно true, но необходимо выставить false. Это должно отключить защиту.

Далее проследуем в папку bin Elasticsearch, откроем там командную строку и введем команду: 

Windows:
Elasticsearch: bin\elasticsearch.bat

macOS/Linux:
Elasticsearch: bin\elasticsearch.bat

После выполнения команды Elasticsearch должен запустится. В консоли вы увидите следующее:

Далее необходимо проверить, работает ли он работает. Для этого открываем браузер и вводим следующий запрос: http://localhost:9200/
Должно появиться примерно это:

Если так, то Elasticsearch запустился, дело за малым.

Запустите Kibana. Для этого нужно открыть папку с этой программной панелью и далее открыть папку bin. После запустите командную строку и введите команду: 

Windows: 
Kibana: bin\kibana.bat

macOS/Linux: 
Kibana: bin\kibana.bat

Когда ваша программа запустится, перейдите по следующей ссылке: http://localhost:5601/

Должна отобразиться такая страница:

После успешной настройки инструментов (Kafka, Elasticsearch, Kibana) можно приступить к подготовке их соединения. Для этого используем скачанный и распакованный logstach. Зайдите в папку config:

Далее необходимо создать файл logstash-kafka.conf и вписать туда следующее:

input {
  kafka {
    bootstrap_servers => "localhost:9092"
    topics => ["test-topic"]
    codec => "json"
  }
}

filter {
  mutate {
    rename => { "name" => "full_name" }
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "test-index"
  }
  stdout { codec => rubydebug } # Вывод в консоль для отладки

Если вы ранее отключили Kafka, Elasticsearch и Kibana, то вот полный план, как всё запустить. Если не отключали, то некоторые пункты (1-2) сможете пропустить: 

1. Запустите ZooKeeper и Apache Kafka

Windows: 

ZooKeeper: bin\windows\zookeeper‑server‑start.bat config\zookeeper.properties

Kafka: bin\windows\kafka‑server‑start.bat config\server.properties

macOS/Linux: 

ZooKeeper: bin/zookeeper‑server‑start.sh config/zookeeper.properties

Kafka: bin/kafka‑server‑start.sh config/server.properties

2. Запустите Elasticsearch и Kibana

Windows: 

Elasticsearch: bin\elasticsearch.bat

Kibana: bin\kibana.bat

macOS/Linux: 

Elasticsearch: bin\elasticsearch.bat

Kibana: bin\kibana.bat

3. Запустите Logstash

Windows: 

logstash: bin\logstash.bat ‑f config\logstash.conf

macOS/Linux: 

logstash: bin/logstash ‑f config/logstash.conf

4. Запустите продюсера Kafka и отправьте сообщение:

Windows: 

Kafka продюсер: bin\windows\kafka‑console‑producer.bat ‑broker‑list localhost:9092 ‑topic logs

macOS/Linux: 

Kafka продюсер: bin/kafka‑console‑producer.sh ‑broker‑list localhost:9092 ‑topic logs

Сообщение (в консоли необходимо его ввести сразу после активации продюсера ): {«name»: «John», «age»: 30, «city»: «New York»}. Просто скопируйте и вставьте его, но в целом вы можете использовать другое сообщение в формате JSON.

5. Открываем Kibana на порту http://localhost:5601/ 

Во вкладке меню, которое располагается слева, ищем пункт Stack Management и нажимаем на него. В открывшемся окне ищем в расположенном справа меню data views и нажимаем на него.

6. Далее ищем кнопку с названием Create data views и нажимаем на неё:

7. В открывшемся окне введите данные и сохраните их:

8. Перейдите в Discover и выберите своё представление.
Чтобы это сделать нажмите на меню и выберите вкладку. А после сделайте так: 

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

Я надеюсь, у вас всё получилось! Однако хочу вам предложить второй способ запуска через Docker. Для этого вам потребуется сам Docker, а так же VPN. К сожалению, Elasticsearch ограничивает возможность скачивать свои инструменты в российском регионе даже через Docker. 

Вот инструкция как установить Docker: Самый простой, полный и понятный туториал Docker для новичков. 

Далее вам необходимо создать 2 файла: 

docker‑compose.yaml 

logstash.conf

Важно! Создать эти файлы лучше всего в одной папке, а саму папку (как и весь путь к файлам) лучше именовать латинскими буквами.

В зависимости от вашей операционной системы, вы можете создать эти файлы следующим образом: 

Windows:

• Откройте Командную строку (cmd).

• Выполните команду:
  echo > myfile.yaml

Mac:

• Откройте Терминал (Cmd + Пробел, затем введите Terminal).

• Выполните команду:
  touch myfile.yaml

Linux:

• Откройте терминал (Ctrl + Alt + T).

• Используйте команду:
  touch myfile.yaml

После создания файлов, вы должны добавить туда следующее содержимое: 

docker-compose.yml

version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:latest
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      kafka_BROKER_ID: 1
      kafka_ZOOKEEPER_CONNECT: "zookeeper:2181"
      kafka_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      kafka_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.17.0
    container_name: elasticsearch
    environment:
      - discovery.type=single-node
      - bootstrap.memory_lock=true
      - ES_JAVA_OPTS=-Xms512m -Xmx512m
      - xpack.security.enabled=false
      - http.port=9200
    ports:
      - "9200:9200"

  logstash:
    image: docker.elastic.co/logstash/logstash:7.17.0
    container_name: logstash
    depends_on:
      - kafka
      - elasticsearch
    ports:
      - "5044:5044"
      - "9600:9600"
    volumes:
      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf
    environment:
      - LS_JAVA_OPTS=-Xmx256m -Xms256m

  kibana:
    image: docker.elastic.co/kibana/kibana:7.17.0
    container_name: kibana
    depends_on:
      - elasticsearch
    ports:
      - "5601:5601"
    environment:
      - elasticsearch_HOSTS=http://elasticsearch:9200
      - XPACK_SECURITY_ENABLED=false

logstash.conf

input {
  kafka {
    bootstrap_servers => "kafka:9092"
    topics => ["logs"]
    codec => "json"
  }
}

output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "kafka-logs-%{+YYYY.MM.dd}"
    ssl => false
  }
  stdout { codec => rubydebug }
}

Далее вам необходимо запустить docker‑compose.yml, используя команду: 

docker‑compose up ‑d

Запустить её вы можете, используя 

CMD или PowerShell (Windows)

Bash (MacOS и Linux)

Таким же образом запускаем Kafka‑console‑producer при помощи следующей команды: 

docker exec ‑it kafka kafka‑console‑producer ‑broker‑list kafka:9092 ‑topic logs

После чего вводим какое‑либо сообщение, например: 

{“timestamp”: “2025-03-14T15:00:00Z”, “level”: “INFO”, “message”: “Hello from kafka!”}

А далее смотрим результат в Kibana

Заключение
В этой статье я показал один из способов интеграции Kafka и Elasticsearch с использованием Logstash в качестве брокера и Kibana для визуализации данных. Теперь у вас есть инструкция, как вручную запустить все компоненты, это позволяет глубже понять принципы их взаимодействия. Однако на практике чаще всего используются контейнеризированные решения, например, с Docker и Docker Compose. Это значительно упрощает развертывание и управление системой.

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

Спасибо за внимание!

Больше авторских материалов для SDET‑специалистов от моих коллег читайте в соцсетях SimbirSoft — ВКонтакте и Telegram.