Описание тега pykafka

PyKafka - это клиент Kafka>=0.8.2 с поддержкой кластеров для Python
0 ответов

Пикафка 2.3.1 Производитель Hang

Мы используем pykafka 2.3.1 и посмотреть, как продюсер зависает время от времени. Сообщение об ошибке: Сообщение не доставлено!! UnknownTopicOrPartition("Произошел сбой при получении запроса на имя темы / от 0 до {ip: порт} с кодом ошибки 3". Мы исп…
01 сен '18 в 13:09
0 ответов

Почему мой продюсер останавливается... может "произвести" только одно сообщение

Я написал программу продюсера pykafka. Программа запрашивает базу данных и вызывает метод для создания сообщения для kafka. Проблема в том, что первое сообщение отправляется нормально. Однако, когда он пытается отправить второе сообщение, я получаю …
09 апр '18 в 19:49
0 ответов

Кафка может получить доступ к списку тем, но не может создавать никаких сообщений

У меня есть один узел Kafka-2.11, установленный в экземпляре EC2, группа безопасности настроена правильно, порты доступны с моего локального компьютера с помощью команды telnet. from pykafka import KafkaClient kafka_host = 'X.X.X.X:9092' client = Ka…
0 ответов

Почему мой потребитель apache kafka случайно игнорирует сообщения в очереди?

Это, вероятно, eisenbug, поэтому я не жду жестких ответов, но больше подсказок о том, что искать, чтобы иметь возможность повторить ошибку. У меня есть управляемая событиями, основанная на Кафке система, состоящая из нескольких сервисов. На данный м…
07 фев '19 в 10:58
1 ответ

Использование PyKafka Api

Я новичок в Kafka и PyKafka. Я знаю, что производитель и потребитель сделаны в PyKafka с помощью приведенного ниже кода. from pykafka import KafkaClient client = KafkaClient("localhost:9092") topic = client.topics["topicname"] producer = topic.get_p…
21 фев '19 в 04:55
1 ответ

Несколько тем и их приоритет

Я использую pykafka для потребления сообщения, и теперь я используюалансированный_консумер для потребления сообщения из одной темы. Теперь я должен потреблять сообщения из другой темы, и если это возможно, приоритетное потребление сообщений из разны…
0 ответов

pykafka: невозможно получить сообщения

Так что я впервые в Kafka и пробую разные Python-библиотеки. Итак, я пытаюсь следующий код: from pykafka import KafkaClient if __name__ == '__main__': client = KafkaClient(hosts="0.0.0.0:9092") print(client.topics) topic = client.topics[b'DAILY_SALE…
14 май '18 в 06:43
1 ответ

Async не работает при использовании pykafka и asyncio

Я пытаюсь вызвать несколько потребительских функций pykafka, используя async. Однако первая функция потребителя pykafka заблокирует работу другой функции. QueueConsumer lib: import json from pykafka import KafkaClient import configparser import asyn…
25 янв '18 в 08:01
0 ответов

Кафка долгое время загрузки координатора и маленькие ISR

Я использую Kafka 0.8.2.1, использую тему с 200 разделами и RF=3, с сохранением журнала в 1GB. Неизвестное событие привело к переходу кластера в состояние "загрузка координатора" или "групповая загрузка". Несколько сигналов сделали это очевидным: по…
22 авг '18 в 22:11
0 ответов

Производитель не может подключиться к брокеру через DNS

У меня есть психический сервер, на котором я установил в adsised.host.name значение ip сервера и выполняю переадресацию портов на маршрутизаторе. Но производитель не может подключиться к брокеру с помощью DNS. ОШИБКА: pykafka.connection: не удалось …
1 ответ

Пикафка - отправка сообщений и получение подтверждений асинхронно

PyKafka имеет ограничение, которое: очередь отчетов о доставке является локальной для потока: она будет обслуживать отчеты только для сообщений, которые были созданы из текущего потока Я пытаюсь написать скрипт, в котором я могу асинхронно отправлят…
0 ответов

Как подключить Кафку к керберизованной среде с помощью Python?

Необходимо подключиться и отправить сообщения из Kafka Client(windows) в среду Kerberized на платформе Hortonworks с использованием python.У нас есть сертификат клиента cacerts который в подстановочный знак с записью *.localdomain,Ниже опробованы во…
06 авг '18 в 09:51
1 ответ

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

Я написал простой продюсер, используя pykafka, но не могу заставить его работать. Основной производитель и призыв к производству ниже. Когда я вызываю это 100 раз с небольшим сообщением и добавляю код синхронизации / профилирования, это занимает око…
02 апр '18 в 18:08
3 ответа

Почему я получил ошибки PartitionOwnedError и ConsumerStoppedException при запуске нескольких потребителей

Я использую pykafka, чтобы получить сообщение из темы kafka, а затем выполнить некоторый процесс и обновить его до mongodb. Поскольку pymongodb может обновлять только один элемент каждый раз, я запускаю 100 процессов. Но при запуске некоторые процес…
21 сен '16 в 11:36
1 ответ

Роль потребителя кафки, отдельного сервиса или компонента Django?

Я разрабатываю аналитику веб-журнала. И я нашел архитектора с Django(Back-end & front-end)+ kafka + spark. Я также нашел такую ​​же систему по этой ссылке: http://thevivekpandey.github.io/posts/2017-09-19-high-velocity-data-ingestion.html с нижеуказ…
4 ответа

Как читать сообщения от группы потребителей kafka без потребления?

Я управляю очередью kafka, используя общую группу потребителей на нескольких машинах. Теперь мне также нужно показать текущее содержимое очереди. Как мне прочитать только те сообщения в группе, которые еще не были прочитаны, но сделать эти сообщения…
21 июн '18 в 09:47
1 ответ

В чем причина TypeError: неразрешимый тип: 'TopicAndPartition', когда KafkaUtils.createDirectStream?

Я хочу использовать сообщение Кафки от любого произвольного смещения на KafkaUtils.createDirectStream, Мой исходный код: import sys from pyspark import SparkContext from pyspark.streaming import StreamingContext from pyspark.streaming.kafka import K…
26 май '16 в 07:25
0 ответов

Как узнать распределение раздела кафки

Я использую kafka 0.9.0 с pykafka в качестве очереди URL в scrapy, и у темы, которую я использовал, было 30 разделов с 4 потребителями. Сначала он работает правильно, но когда он работал около двух часов, у кафки, похоже, возникла проблема с перебал…
24 апр '17 в 08:45
1 ответ

PyKafka не может подключиться к разделенной теме кафки

apache-storm-1.0.2 hbase-1.2.1 kafka_2.10-0.10.0.0 zookeeper-3.4.9 опс: zkServer.sh start $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties &! $KAFKA_HOME/bin/kafka-topics.sh --zookeeper localhost:2181 --create --replica…
2 ответа

Как создать тему кафки, используя пакет сообщений или буфер с pykafka

Как создать тему кафки используя пакет сообщений или буфер с пикафкой. Я имею в виду, что один производитель может создать много сообщений в одном процессе. Я знаю концепцию, используя сообщение пакетного или буферного сообщения, но я не знаю, как е…