Write to Kafka from Dask Stream
Nobody has claimed this yet.
Assessment
- Difficulty
- 5/5
- Estimated time
- Over a week
- Newbie friendliness
- 25/100
- Issue type
- Feature
- Clarity
- Needs clarification
- Activity status
- Stale
- Tech stack
- kafka, python
- Domain
- distributed-systems, stream-processing
Research direction
Start by reviewing the existing DaskStream and core.to_kafka implementations, including to_kafka.update and the get_producer workaround shown in the issue. Determine how the existing back-pressure handling and worker-local Kafka connection should fit together without pickling the producer. Done means Dask Stream can write to Kafka with the existing back-pressure behavior and a reliable worker-side producer.
Written by the indexing model from the issue text.
Description
It seems like the exiting PR for this needs improvements. I'm creating this issue to invite discussion on now to properly solve this problem.
I would like to use the existing back pressure handling implementation in the existing to_kafka function. I'm thinking that we can just call the existing to_kafka.update from Dask. It looks something like this:
@DaskStream.register_api()
class to_kafka(DaskStream, core.to_kafka):
def update(self, x, who=None, metadata=None):
client = default_client()
result = client.submit(to_kafka.update, self, x, who, metadata)
yield self._emit(result)
However, there's still the issue of the Kafka connection. It seems Dask attempts to pickle the Kafka producer and push it the worker. This fails due to the Confluent producer using some sort of lazy instantiation. In my previous attempt to make this work, I was able to create the producer using the getattr workaround. It also attaches the producer to the worker in order to maintain the connection.
def get_producer(config):
w = get_worker()
if hasattr(w, 'producer'):
return w.producer
import confluent_kafka as ck
w.producer = getattr(ck, 'Producer')(config)
return w.producer
This seems very much like a hack to me, and I would prefer a better solution. If anyone has more expertise on Dask, any help would be appreciated.
- Dominant language
- Python
- Stars
- 1.3k
- Forks
- 149
- Avg merge
- 17h 39m
- Merged PRs (30d)
- 1
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
More from python-streamz/streamz
-
Difficulty 3/5 1-2 days Newbie friendliness 35/100
python-streamz/streamz#481 · 4 comments ·
-
Combining the streamz.Stream.filenames() and streamz.Stream.from_textfile() using dask scatter? Open
Difficulty 5/5 Over a week Newbie friendliness 25/100
python-streamz/streamz#480 · 2 comments ·
-
Difficulty 5/5 Over a week Newbie friendliness 20/100
python-streamz/streamz#479 · 1 comment ·
-
Difficulty 5/5 Over a week Newbie friendliness 20/100
python-streamz/streamz#478 · 6 comments ·
-
Difficulty 4/5 3-5 days Newbie friendliness 10/100
python-streamz/streamz#476 · 17 comments · 2 reactions ·
All issues in python-streamz/streamz
Similar issues
-
Add: hunch Open
Difficulty 2/5 1-3 hours Newbie friendliness 74/100
AbdelStark/awesome-typesafe#104 ·
-
enhancement
Difficulty 2/5 1-3 hours Newbie friendliness 72/100
-
Difficulty 2/5 1-3 hours Newbie friendliness 68/100
DiamondLightSource/dodal#2211 ·
-
Difficulty 2/5 1-3 hours Newbie friendliness 88/100
openml/openml-python#1749 ·
-
Difficulty 2/5 1-3 hours Newbie friendliness 88/100
sipyourdrink-ltd/bernstein#6191 ·