<method-wrapper '__call__' of functools.partial object at 0x1356e10> is not a Python function

I'm trying to build a function which I can use as a handler for an RxPy stream that I'm mapping over. The function I have needs access to a variable outside the scope where that variable is...

Use Cases for Kafka Connect

I have been learning about Kafka lately, in order to use the framework to send, receive, and store messages for our worker processes, using the python client (pykafka). Our messages are fairly...

Producer can not connect to broker through DNS

I have a physical server, where i have set the advertised.host.name to server ip, and do the port forwarding at the router. But the producer cannot connect to broker by using the...

Kafka and Pyspark Integration

I am naive in Big data, I am trying to connect kafka to spark. Here is my producer code import os import sys import pykafka def get_text(): ## This block generates my required text. ...

kafka streaming or spark streaming

Am using now kafka in Python. Was wondering if Spark Kafka is needed or can we use just use kafka through pyKafka. My concern was Spark creates overhead (pyspark) in the process, and if we don't...

How to read messages from kafka consumer group without consuming?

I'm managing a kafka queue using a common consumer group across multiple machines. Now I also need to show the current content of the queue. How do I read only those messages within the group...

Kafka long coordinator load time and small ISRs

I'm using Kafka 0.8.2.1, running a topic with 200 partitions and RF=3, with log retention set to about 1GB. An unknown event caused the cluster to enter the "coordinator load" or "group load"...

Pykafka - sending messages and receiving acknowledgments asynchronously

PyKafka has the limitation that: delivery report queue is thread-local: it will only serve reports for messages which were produced from the current thread I'm trying to write a script where I...

Multiple topics and priority of them

I am using pykafka for consuming message and now I am using balanced_consumer for consuming message from one topic. Now I have to consume messages from another topic, and if it is possible to...

Kafka Producer and Consumer Scripts to Run automatically

I have a Django project and I am using pykafka. I have created two files named producer.py and consumer.py inside the project. I have to change directory into the folder where these are present...

Is there a Python API for event-driven Kafka consumer?

I have been trying to build a Flask app that has Kafka as the only interface. For this reason, I want have a Kafka consumer that is triggered when there is new message in the stream of the...

pykafka queued.max.messages.kbytes is outside allowed range

I am trying to use pykafka with rdkafka to consume messages but I am getting the following error while trying to initialize the consumer. Configuration property "queued.max.messages.kbytes" value...

Why does my apache kafka consumer randomly ignores queued messages?

This is probably an eisenbug so I'm not expecting hard answers but more hints on what to look for to be able to replicate the bug. I have an event-driven, kafka-based system composed of several...

PyKafka Api usage

I am a newbie to Kafka and PyKafka.I know that a producer and a consumer are made in PyKafka via the below code. from pykafka import KafkaClient client = KafkaClient("localhost:9092") topic =...

Produce/Consume to Remote Kafka Does not Work

I have set up a AWS EC2 instance running Apache Kafka 0.8 via a Bitnami AMI image. The server properties are pretty much default (Kafka located at localhost:9092 and zookeeper located at...

How to work with kafka/pykafka balanced consumers?

I'm entirely new to kafka. I got to work with simple consumer, and it was ok. Now i've got work with balanced consumer, the problem is i have only a vague idea of kafka in general. topic =...

TypeError: produce() got an unexpected keyword argument 'linger_ms'

I'm trying to use kafka with python with pykafka , when i try to use linger_ms i get this error : TypeError: produce() got an unexpected keyword argument 'linger_ms' This is my code : import...

Not able to communicate from a remote machine to Kafka cluster

So i have been trying to send messages using pykafka on non kafka cluster (it doesn't have kafka on it, only necessary libraries) Using the attached code snippet i send message to kafka cluster...

pip install uwsgi failed with /usr/local/include/string/string.h:7:10: fatal error: 'sstream' file not found?

I am trying to install uwsgi in virtualenv on macOS-10.14.5 with python2.7. I am getting this error /usr/local/include/string/string.h:7:10: fatal error: 'sstream' file not found I tried updating...

kafka ssl for python and java config

I want to build ssl config for python java build jks for java bild pem for python I merge them in one...

Can't establish SSL connection to Kafka after upgrading to python 3.7

Code I have that successfully connects to Kafka with an SSL connection in Python 3.6.7 fails when using Python 3.7.3, with error message SSL: WRONG_VERSION_NUMBER. I would not expect code working...

Cant send message by using pykafka

I am using pykafka, I can get topic names but I cant send message. My code is shown below client = KafkaClient(hosts='xx.xx.xx.xx:9092') topic = client.topics['test'] producer =...

Pytest a KAFKA consumer and producer (integration test)

I am new to Pytest framework. I currently have a Python app which consumes and produces messages using Kafka-Python. I am trying to write integration tests which would verify if the has been...

How to setup Kafka Client within Google App Engine

I have followed the google tutorial below and am able to run this on my google app engine no problem. https://cloud.google.com/appengine/docs/standard/python3/building-app/writing-web-service I...

Write a csv file to a kafka topic

I have a large csv and I want to write to a kafka topic. def producer(): producer = KafkaProducer(bootstrap_servers='mykafka-broker') with open('/home/antonis/repos/testfile.csv') as...

from kafka import KafkaClient ImportError: No module named kafka

I installed Python version 3.8.3 and 2.7.16, pip install pykafka, and pip install kafka-python. When I try to run the Python program with installed modules I get an error: from kafka import...

I need to create a docker-compose for flask and several other libraries?

I need to create a docker-compose file for flask based API back end application. But this application is based on several other libraries following. All need to run on docker containers only. Can...

Twitter+Apache Kafka+ Spark Structured Streaming doesn't work

I want to work some example codes from github(https://github.com/kaantas/spark-twitter-sentiment-analysis). I follow steps below; Started zkserver Started kafka 2.5.0 version (also i am using...

Kafka reading txt file as single letters rather than words

I'm using the KafkaClient in python's pykafka. I'm trying to read a text file and produce its lines to a topic then read it by a consumer. However on running it only reads individual letters in...

pip3.6 install mysqlclient==1.3.12 fails with error: unknown type name ‘my_bool’; did you mean ‘bool

I have a project that worked on ubuntu 16.04 with python 3.6 but now we are trying to make it run on ubuntu 20.04 with same python version. I need to install all requirements on the venv and...