Rivalry2

HTML Diff

1 added 1 removed

Original 2026-01-01

Modified 2026-02-19

1 Apache Kafka® лежит в основе большой экосистемы, куда входят такие мощные компоненты, как Kafka Connect и Kafka Streams. В этой экосистеме для разработчиков есть и другие инструменты. С ними работа станет проще и приятнее. Рассмотрим несколько таких инструментов.<p>Материал подготовлен на основе перевода<a>оригинальной статьи в блоге Confluent</a>, автор - <a>Dave Klein</a>.</p>

2 <strong>Подробнее об инструментах Кафки вы сможете узнать на курсе</strong><a><strong>"Apache Kafka База"</strong></a><strong>от Слёрма. 6-недельный курс заменит месяцы самостоятельного погружения в тему по книгам и технической документации.</strong><h2>kcat </h2>

3 Начнём с этого потрясающего инструмента.<a>kcat</a>(ранее - kafkacat) - это быстрая и гибкая утилита командной строки для работы с топиками (создания, удаления), отправки и получения данных в/из Kafka и т. д. Разработчик - Магнус Эденхилл (Magnus Edenhill), создатель библиотеки librdkafka C/C++ для Kafka. Это отличный инструмент для быстрого создания и потребления данных из топика. Одна команда может делать и то, и другое, в зависимости от контекста. Например:<blockquote>$ ~ echo "Hello World" | kafkacat -b localhost:29092 -t hello-topic</blockquote><blockquote>% Auto-selecting Producer mode (use -P or -C to override)</blockquote>Мы отправили данные в stdout с помощью echo и передали их в kcat. Нам понадобилось два простых флага: -b для указания брокера и -t для указания названия целевого топика. kcat понимает, что мы посылаем ему данные, и переходит в режим продюсера. Вычитать данные из топика можно точно такой же командой:<blockquote>$ ~ kafkacat -b localhost:29092 -t hello-topic</blockquote><blockquote>% Auto-selecting Consumer mode (use -P or -C to override)</blockquote><blockquote>Hello World</blockquote><blockquote>% Reached end of topic hello-topic [0] at offset 1</blockquote>Если мы хотим отправить запись с ключом, нужно просто использовать разделитель и флаг -K. В этом случае используем двоеточие:<blockquote>$ ~ echo "123:Jane Smith" | kafkacat -b localhost:29092 -t customers -K:</blockquote><blockquote>% Auto-selecting Producer mode (use -P or -C to override)</blockquote>И снова та же команда считает запись из топика:<blockquote>$ ~ kafkacat -b localhost:29092 -t customers -K:</blockquote><blockquote>% Auto-selecting Consumer mode (use -P or -C to override)</blockquote><blockquote>123:Jane Smith</blockquote><blockquote>% Reached end of topic customers [0] at offset 1</blockquote>Можно не указывать флаг -K при чтении, если нам требуется только значение:<blockquote>$ ~ kafkacat -b localhost:29092 -t customers</blockquote><blockquote>% Auto-selecting Consumer mode (use -P or -C to override)</blockquote><blockquote>Jane Smith</blockquote><blockquote>% Reached end of topic customers [0] at offset 1</blockquote>Передавая данные из stdout в kcat, мы запускаем продюсер, отправляем данные и закрываем продюсер. Если мы хотим, чтобы продюсер продолжил работу, указываем флаг -P, как указано в подсказке auto-selecting.<p>Консьюмер продолжит работу, как сделал бы kafka-console-consumer. Чтобы получить данные из топика и сразу выйти, используем флаг -e.</p>

4 <p>Чтобы получить данные в формате Avro, используем флаг -s: для всей записи (-s avro), для ключа (-s key=avro) или только для значения (-s value=avro). Вот пример использования топика movies из популярного</p>

5 <a>руководства по оценке фильмов</a>:<blockquote>$ ~ kafkacat -C -b localhost:29092 -t movies -s value=avro -r http://localhost:8081</blockquote><blockquote>------------------------------------------------------</blockquote><blockquote>{"id": 294, "title": "Die Hard", "release_year": 1988}</blockquote><blockquote>{"id": 354, "title": "Tree of Life", "release_year": 2011}</blockquote><blockquote>{"id": 782, "title": "A Walk in the Clouds", "release_year": 1995}</blockquote><blockquote>{"id": 128, "title": "The Big Lebowski", "release_year": 1998}</blockquote><blockquote>{"id": 780, "title": "Super Mario Bros.", "release_year": 1993}</blockquote>Это на самом деле мощный инструмент, флагов там намного больше. Запустите kafkacat -h, чтобы получить полный список. Больше примеров использования kcat читайте в<a>блоге Robin Moffatt</a>.<p>В kcat недостаёт<a>возможности создавать данные в формате</a> Avro. Мы уже видели, что получать Avro можно с помощью Confluent Schema Registry, а отправлять нельзя. Поэтому переходим к следующему инструменту.</p>

6 <h2>Confluent REST Proxy</h2>

7 <a>Confluent REST Proxy</a> - это клиент HTTP Kafka со множеством функций. С его помощью можно предоставлять поддержку Kafka в приложениях на языках, для которых нет нативного клиента Kafka. Это его основное применение, но есть и другие. Например, он создаёт данные Avro в топике Kafka:<blockquote>$ ~ curl -X POST \</blockquote><blockquote>-H "Content-Type: application/vnd.kafka.avro.v2+json" \</blockquote><blockquote>-H "Accept: application/vnd.kafka.v2+json" \</blockquote><blockquote>--data @newMovieData.json "http://localhost:8082/topics/movies"</blockquote><blockquote>------------------------------------------------------------------------------</blockquote><blockquote>{"offsets":[{"partition":0,"offset":5,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":3}</blockquote>REST Proxy входит в Confluent Platform и предоставляется по лицензии Confluent Community License, но может использоваться отдельно с любым кластером Kafka. Он действительно на многое способен -<a>читайте документацию</a>.<p>Как видите, REST Proxy можно использовать из командной строки с curl или чем-то подобным. Ещё его можно использовать с такими инструментами, как Postman, для создания удобного пользовательского интерфейса Kafka.</p>

8 <p>Пример создания данных в топике с помощью Postman (заголовки Content-Type и Accept заданы на вкладке Headers):</p>

9 Как мы видели в примерах с curl и Postman, REST Proxy требует передавать схему для сообщений Avro с каждым запросом на создание. С Postman или подобным инструментом, который позволяет создавать библиотеку сохранённых запросов, этими процессами будет проще управлять.<p>Чтобы потреблять данные из топика с помощью REST Proxy, мы сначала создаём консьюмера в группе консьюмеров, а потом подписываемся на топик (или топики) и, наконец, получаем записи. Переключаемся обратно на curl, так мы сможем увидеть все необходимые элементы сразу.</p>

10 <p>Сначала выполняем POST эндпоинте консьюмера, указав имя нашей группы консьюмеров. В этом запросе POST мы передадим имя нового инстанса консьюмера, формат внутренних данных (у нас это Avro) и значение auto.offset.reset.</p>

11 <blockquote>$ ~ curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" \</blockquote><blockquote> --data '{"name": "movie_consumer_instance", "format": "avro", "auto.offset.reset": "earliest"}' \</blockquote><blockquote> http://localhost:8082/consumers/movie_consumers</blockquote><blockquote>------------------------------------------------------------------------------</blockquote><blockquote>{"instance_id":"movie_consumer_instance","base_uri":"http://localhost:8082/consumers/movie_consumers/instances/movie_consumer_instance"}</blockquote>Вернётся instance id и базовый URI для только что созданного инстанса консьюмера. Укажем этот URI, чтобы подписаться на топик, указав конечную точку subscription.<blockquote>$ ~ curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" --data '{"topics":["movies"]}' \ http://localhost:8082/consumers/movie_consumers/instances/movie_consumer_instance/subscription</blockquote>Вернётся только ответ 204. Теперь можно выполнить запрос GET эндпоинту records по тому же URI, чтобы получить записи.<blockquote>$ ~ curl -X GET -H "Accept: application/vnd.kafka.avro.v2+json" \ http://localhost:8082/consumers/movie_consumers/instances/movie_consumer_instance/records</blockquote><blockquote>----------------------------------------------------------------------</blockquote><blockquote>[{"topic":"movies","key":null,"value":{"id": 294, "title": "Die Hard", "release_year": 1988},"partition":0,"offset":0},</blockquote><blockquote>{"topic":"movies","key":null,"value":{"id": 354, "title": "Tree of Life", "release_year": 2011},"partition":0,"offset":1},{"topic":"movies","key":null,"value":{"id": 782, "title": "A Walk in the Clouds", "release_year": 1995},"partition":0,"offset":2},{"topic":"movies","key":null,"value":{"id": 128, "title": "The Big Lebowski", "release_year": 1998},"partition":0,"offset":3},{"topic":"movies","key":null,"value":{"id": 780, "title": "Super Mario Bros.", "release_year": 1993},"partition":0,"offset":4},{"topic":"movies","key":null,"value":{"id":101,"title":"Chariots of Fire","release_year":1981},"partition":0,"offset":5}]</blockquote>Созданный консьюмер останется, так что мы сможем выполнить тот же запрос GET в любое время, чтобы проверить новые данные. Если этот консьюмер больше не нужен, можно его удалить с помощью DELETE и базового URI.<blockquote>$ ~ curl -X DELETE -H "Content-Type: application/vnd.kafka.v2+json" \ http://localhost:8082/consumers/movie_consumers/instances/movie_consumer_instance</blockquote>Простые запросы GET позволяют получать информацию о brokers, topics и partitions.<blockquote>$ ~ curl "http://localhost:8082/brokers"</blockquote><blockquote>$ ~ curl "http://localhost:8082/topics"</blockquote><blockquote>$ ~ curl "http://localhost:8082/topics/movies"</blockquote><blockquote>$ ~ curl "http://localhost:8082/topics/movies/partitions"</blockquote>Эти запросы возвращают много данных JSON, так что здесь мы их приводить не будем, чтобы не занимать место. И это подводит нас к следующему инструменту.<h2>jq: утилита для работы с JSON из командной строки</h2>

12 jq используется не только с Kafka. Это очень полезный инструмент, если мы работаем с другими утилитами командной строки, которые возвращают данные JSON. С помощью jq можно форматировать, обрабатывать и извлекать данные из выходных данных других программ в формате JSON. Инструкции по загрузке и установке jq см.<a>на GitHub</a>. Там же вы найдете ссылки на инструкции и другие ресурсы.<p>Вернёмся к выходным данным REST Proxy в запросе GET к консьюмеру. Это не самый большой BLOB в формате JSON, но всё же читать его неудобно. Попробуем ещё раз, на этот раз передав выходные данные в jq:</p>

13 <blockquote>$ ~ curl -X GET -H "Accept: application/vnd.kafka.avro.v2+json" \</blockquote><blockquote> http://localhost:8082/consumers/movie_consumers/instances/movie_consumer_instance/records | jq</blockquote><blockquote> </blockquote><blockquote>[</blockquote><blockquote> {</blockquote><blockquote> "topic": "movies",</blockquote><blockquote> "key": null,</blockquote><blockquote> "value": {</blockquote><blockquote> "id": 294,</blockquote><blockquote> "title": "Die Hard",</blockquote><blockquote> "release_year": 1988</blockquote><blockquote> },</blockquote><blockquote> "partition": 0,</blockquote><blockquote> "offset": 0</blockquote><blockquote> },</blockquote><blockquote> {</blockquote><blockquote> "topic": "movies",</blockquote><blockquote> "key": null,</blockquote><blockquote> "value": {</blockquote><blockquote> "id": 354,</blockquote><blockquote> "title": "Tree of Life",</blockquote><blockquote> "release_year": 2011</blockquote><blockquote> },</blockquote><blockquote> "partition": 0,</blockquote><blockquote> "offset": 1</blockquote><blockquote> },</blockquote><blockquote> {</blockquote><blockquote> "topic": "movies",</blockquote><blockquote> "key": null,</blockquote><blockquote> "value": {</blockquote><blockquote> "id": 782,</blockquote><blockquote> "title": "A Walk in the Clouds",</blockquote><blockquote> "release_year": 1995</blockquote><blockquote> },</blockquote><blockquote> "partition": 0,</blockquote><blockquote> "offset": 2</blockquote><blockquote> },</blockquote><blockquote> {</blockquote><blockquote> "topic": "movies",</blockquote><blockquote> "key": null,</blockquote><blockquote> "value": {</blockquote><blockquote> "id": 128,</blockquote><blockquote> "title": "The Big Lebowski",</blockquote><blockquote> "release_year": 1998</blockquote><blockquote> },</blockquote><blockquote> "partition": 0,</blockquote><blockquote> "offset": 3</blockquote><blockquote> },</blockquote><blockquote> {</blockquote><blockquote> "topic": "movies",</blockquote><blockquote> "key": null,</blockquote><blockquote> "value": {</blockquote><blockquote> "id": 780,</blockquote><blockquote> "title": "Super Mario Bros.",</blockquote><blockquote> "release_year": 1993</blockquote><blockquote> },</blockquote><blockquote> "partition": 0,</blockquote><blockquote> "offset": 4</blockquote><blockquote> },</blockquote><blockquote> {</blockquote><blockquote> "topic": "movies",</blockquote><blockquote> "key": null,</blockquote><blockquote> "value": {</blockquote><blockquote> "id": 101,</blockquote><blockquote> "title": "Chariots of Fire",</blockquote><blockquote> "release_year": 1981</blockquote><blockquote> },</blockquote><blockquote> "partition": 0,</blockquote><blockquote> "offset": 5</blockquote><blockquote> }</blockquote><blockquote>]</blockquote>Так понятнее, но всё равно много лишнего. Допустим, мы хотим видеть только названия фильмов и годы выпуска. Это легко сделать с помощью jq:<blockquote>$ ~ curl -X GET -H "Accept: application/vnd.kafka.avro.v2+json" \</blockquote><blockquote> http://localhost:8082/consumers/movie_consumers/instances/movie_consumer_instance/records | jq \</blockquote><blockquote> | jq '.[] | {title: .value.title, year: .value.release_year}'</blockquote><blockquote> </blockquote><blockquote>{</blockquote><blockquote> "title": "Die Hard",</blockquote><blockquote> "year": 1988</blockquote><blockquote>}</blockquote><blockquote>{</blockquote><blockquote> "title": "Tree of Life",</blockquote><blockquote> "year": 2011</blockquote><blockquote>}</blockquote><blockquote>{</blockquote><blockquote> "title": "A Walk in the Clouds",</blockquote><blockquote> "year": 1995</blockquote><blockquote>}</blockquote><blockquote>{</blockquote><blockquote> "title": "The Big Lebowski",</blockquote><blockquote> "year": 1998</blockquote><blockquote>}</blockquote><blockquote>{</blockquote><blockquote> "title": "Super Mario Bros.",</blockquote><blockquote> "year": 1993</blockquote><blockquote>}</blockquote><blockquote>{</blockquote><blockquote> "title": "Chariots of Fire",</blockquote><blockquote> "year": 1981</blockquote><blockquote>}</blockquote>Давайте посмотрим, что у нас получилось (можете проделать<a>то же самое в jqplay</a>):<ol><li>Мы передали выходные данные из REST Proxy в jq. В одинарных кавычках выполняется программа jq с двумя шагами. jq использует вертикальную черту, чтобы передать выходные данные из одного шага как входные для другого.</li>

14 <li>В нашем примере первый шаг - это итератор, который считывает запись каждого фильма из массива и передаёт её на следующий шаг.</li>

15 <li>На втором шаге создаётся новый объект JSON из каждой записи. Ключи произвольные, а значения берутся из выходных данных с помощью оператора identity, '.'.</li>

16 </ol>С jq можно делать много интересного. Изучите<a>документацию</a>.<p>Обработка данных JSON в jq позволяет нам сочетать разные операции для достижения желаемого результата. Это напоминает мне о Kafka Streams и нашем последнем инструменте.</p>

17 <h2>Kafka Streams Topology Visualizer</h2>

18 <a>Kafka Streams Topology Visualizer</a>обрабатывает текстовое описание топологии Kafka Streams и графически изображает входные топики, обрабатывающие ноды, промежуточные топики, хранилища состояний и т. д. Это отличный способ получить общее представление о сложной топологии Kafka Streams. В руководстве по оценке фильмов топология не такая уж и сложная, но хорошо послужит для иллюстрации.<p>Текстовое представление топологии, полученное с помощью метода Topology::describe:</p>

19 <blockquote>Topologies:</blockquote><blockquote> Sub-topology: 0</blockquote><blockquote> Source: KSTREAM-SOURCE-0000000000 (topics: [movies])</blockquote><blockquote> --> KSTREAM-MAP-0000000001</blockquote><blockquote> Processor: KSTREAM-MAP-0000000001 (stores: [])</blockquote><blockquote> --> KSTREAM-SINK-0000000002</blockquote><blockquote> <-- KSTREAM-SOURCE-0000000000</blockquote><blockquote> Sink: KSTREAM-SINK-0000000002 (topic: rekeyed-movies)</blockquote><blockquote> <-- KSTREAM-MAP-0000000001 Sub-topology: 1 Source: KSTREAM-SOURCE-0000000010 (topics: [KSTREAM-MAP-0000000007-repartition]) --> KSTREAM-JOIN-0000000011</blockquote><blockquote> Processor: KSTREAM-JOIN-0000000011 (stores: [rekeyed-movies-STATE-STORE-0000000003])</blockquote><blockquote> --> KSTREAM-SINK-0000000012</blockquote><blockquote> <-- KSTREAM-SOURCE-0000000010 Source: KSTREAM-SOURCE-0000000004 (topics: [rekeyed-movies]) --> KTABLE-SOURCE-0000000005</blockquote><blockquote> Sink: KSTREAM-SINK-0000000012 (topic: rated-movies)</blockquote><blockquote> <-- KSTREAM-JOIN-0000000011 Processor: KTABLE-SOURCE-0000000005 (stores: [rekeyed-movies-STATE-STORE-0000000003]) --> none</blockquote><blockquote> <-- KSTREAM-SOURCE-0000000004 Sub-topology: 2 Source: KSTREAM-SOURCE-0000000006 (topics: [ratings]) --> KSTREAM-MAP-0000000007</blockquote><blockquote> Processor: KSTREAM-MAP-0000000007 (stores: [])</blockquote><blockquote> --> KSTREAM-FILTER-0000000009</blockquote><blockquote> <-- KSTREAM-SOURCE-0000000006 Processor: KSTREAM-FILTER-0000000009 (stores: []) --> KSTREAM-SINK-0000000008</blockquote><blockquote> <-- KSTREAM-MAP-0000000007</blockquote><blockquote> Sink: KSTREAM-SINK-0000000008 (topic: KSTREAM-MAP-0000000007-repartition)</blockquote><blockquote> <-- KSTREAM-FILTER-0000000009</blockquote>Даже если вам по этому описанию всё понятно, поверьте, большинству людей гораздо удобнее использовать графическую схему.Kafka Streams Topology Visualizer - это веб-приложение, которое можно разместить в своей среде (загрузив с<a>GitHub</a>). Если вы используете его время от времени, достаточно будет<a>общедоступной версии</a>.<p>Сложную топологию неудобно изучать целиком, поэтому можно визуализировать ее частями, а потом объединить изображения. Это особенно удобно, когда новому разработчику нужно быстро изучить существующее приложение Kafka Streams.</p>

20 <h2>Топологии ksqlDB</h2>

21 <a>ksqlDB</a> - это база данных потоковой передачи событий, с помощью которой можно создавать сложные топологии, используя синтаксис, знакомый каждому разработчику SQL. Поскольку ksqlDB создается поверх Kafka Streams, Kafka Streams Topology Visualizer работает и с этими типами топологий.<p>Получаем описание топологии из ksqlDB командой EXPLAIN. Для начала находим исполняющий запрос:</p>

22 - <blockquote>ksql> SHOW QUERIES;</blockquote><blockquote><strong> </strong></blockquote><blockquote> Query ID | Query Type | Status | Sink Name | Sink Kafka Topic | Query String</blockquote><blockquote>--------------------------------------------------------------------------------------------------------------------------</blockquote><blockquote> CSAS_SHIPPED_ORDERS_0 | PERSISTENT | RUNNING:1 | SHIPPED_ORDERS | SHIPPED_ORDERS | CREATE STREAM SHIPPED_ORDERS WITH </blockquote><blockquote> ...</blockquote>Теперь можно использовать созданное имя запроса CSAS_SHIPPED_ORDERS_0 для получения топологии:<blockquote>ksql> EXPLAIN CSAS_SHIPPED_ORDERS_0;</blockquote>Выходные данные будут объемными, ��ак что здесь мы их приводить не будем. Копируем и вставляем описание топологии в визуализатор и получаем такую схему:<h2>Верхушка айсберга</h2>

22 + <blockquote>ksql> SHOW QUERIES;</blockquote><blockquote><strong> </strong></blockquote><blockquote> Query ID | Query Type | Status | Sink Name | Sink Kafka Topic | Query String</blockquote><blockquote>--------------------------------------------------------------------------------------------------------------------------</blockquote><blockquote> CSAS_SHIPPED_ORDERS_0 | PERSISTENT | RUNNING:1 | SHIPPED_ORDERS | SHIPPED_ORDERS | CREATE STREAM SHIPPED_ORDERS WITH </blockquote><blockquote> ...</blockquote>Теперь можно использовать созданное имя запроса CSAS_SHIPPED_ORDERS_0 для получения топологии:<blockquote>ksql> EXPLAIN CSAS_SHIPPED_ORDERS_0;</blockquote>Выходные данные будут объемными, так что здесь мы их приводить не будем. Копируем и вставляем описание топологии в визуализатор и получаем такую схему:<h2>Верхушка айсберга</h2>

23 Мы рассмотрели четыре полезных инструмента для разработчиков Apache Kafka, но их гораздо больше, и это одно из преимуществ работы с таким активным сообществом. Если вы знаете полезный инструмент командной строки или графическое приложение для Kafka, расскажите об этом на форуме<a>Confluent Community Forum</a>.<p>→ <a>Курс "Apache Kafka База"</a></p>