# Pulsar plugin for Streamz
This a plugin for [Streamz](https://github.com/python-streamz/streamz) that adds stream
nodes for writing and reading data from/to Pulsar.
## 🛠 Installation
Latest stable version is available on PyPI
```sh
pip install streamz_pulsar
```
Latest development version can be installed from git repo
```sh
pip install git+https://github.com/MarekWadinger/streamz_pulsar
```
## ⚡️ Quickstart
To start working with streamz_pulsar, follow these 3 steps:
### 1. Run a standalone Pulsar cluster locally
```sh
docker run -it -p 6650:6650 -p 8000:8080 --mount source=pulsardata,target=/pulsar/data --mount source=pulsarconf,target=/pulsar/conf apachepulsar/pulsar:latest bin/pulsar standalone
```
### 2. Create a consumer
The following example creates a consumer with the `my-sub` subscription name on the `my-topic` topic, receives incoming messages, prints the content and ID of messages that arrive, and acknowledges each message to the Pulsar broker.
```python
import pulsar
from streamz import Stream
s = Stream.from_pulsar(
['my-topic'],
subscription_name='my-sub',
consumer_params={'service_url': 'pulsar://localhost:6650'}
)
s.map(lambda x: x.decode())
L = s.sink_to_list()
s.start()
while True:
try:
if L:
print(L.pop(0))
except pulsar.Interrupted:
print("Stop receiving messages")
break
```
### 3. Create a producer
The following example creates a Python producer for the `my-topic` topic and sends 10 messages on that topic:
```python
from streamz import Stream
source = Stream()
producer_ = source.to_pulsar(
'my-topic',
producer_config={'service_url': 'pulsar://localhost:6650'}
)
for i in range(3):
source.emit(('hello-pulsar-%d' % i).encode('utf-8'))
producer_.stop()
producer_.flush()
```
Raw data
{
"_id": null,
"home_page": "https://github.com/MarekWadinger/streamz_pulsar",
"name": "streamz-pulsar",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.7.2,<4.0.0",
"maintainer_email": "",
"keywords": "streamz,pulsar",
"author": "Marek Wadinger",
"author_email": "marekwadinger@icloud.com",
"download_url": "https://files.pythonhosted.org/packages/37/05/306cb32f021c341fc83738ca3dcb65e841cf389ed5a10e8d0e036a15d434/streamz_pulsar-0.1.0.post2.tar.gz",
"platform": null,
"description": "# Pulsar plugin for Streamz\n\nThis a plugin for [Streamz](https://github.com/python-streamz/streamz) that adds stream\nnodes for writing and reading data from/to Pulsar.\n\n## \ud83d\udee0 Installation\n\nLatest stable version is available on PyPI\n\n```sh\npip install streamz_pulsar\n```\n\nLatest development version can be installed from git repo\n\n```sh\npip install git+https://github.com/MarekWadinger/streamz_pulsar\n```\n\n## \u26a1\ufe0f Quickstart\n\nTo start working with streamz_pulsar, follow these 3 steps:\n\n### 1. Run a standalone Pulsar cluster locally\n\n```sh\ndocker run -it -p 6650:6650 -p 8000:8080 --mount source=pulsardata,target=/pulsar/data --mount source=pulsarconf,target=/pulsar/conf apachepulsar/pulsar:latest bin/pulsar standalone\n```\n\n### 2. Create a consumer\n\nThe following example creates a consumer with the `my-sub` subscription name on the `my-topic` topic, receives incoming messages, prints the content and ID of messages that arrive, and acknowledges each message to the Pulsar broker.\n\n```python\nimport pulsar\nfrom streamz import Stream\n\ns = Stream.from_pulsar(\n ['my-topic'],\n subscription_name='my-sub',\n consumer_params={'service_url': 'pulsar://localhost:6650'}\n )\n\ns.map(lambda x: x.decode())\nL = s.sink_to_list()\n\ns.start()\nwhile True:\n try:\n if L:\n print(L.pop(0))\n except pulsar.Interrupted:\n print(\"Stop receiving messages\")\n break\n```\n\n### 3. Create a producer\n\nThe following example creates a Python producer for the `my-topic` topic and sends 10 messages on that topic:\n\n```python\nfrom streamz import Stream\n\nsource = Stream()\nproducer_ = source.to_pulsar(\n 'my-topic',\n producer_config={'service_url': 'pulsar://localhost:6650'}\n )\n\nfor i in range(3):\n source.emit(('hello-pulsar-%d' % i).encode('utf-8'))\n\nproducer_.stop()\nproducer_.flush()\n\n```\n",
"bugtrack_url": null,
"license": "",
"summary": "",
"version": "0.1.0.post2",
"project_urls": {
"Homepage": "https://github.com/MarekWadinger/streamz_pulsar",
"Repository": "https://github.com/MarekWadinger/streamz_pulsar"
},
"split_keywords": [
"streamz",
"pulsar"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "a6d9323365fd272b489e77d19b38e3045a2be47f0aec785f811f1081dd0097b3",
"md5": "63bc6b1bd6d744e7f8c31e780b9a7575",
"sha256": "4051121e01eda711d5a6e1c30f84c6a0e5d06d3d34f0aa9c84ff67a13637eba4"
},
"downloads": -1,
"filename": "streamz_pulsar-0.1.0.post2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "63bc6b1bd6d744e7f8c31e780b9a7575",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7.2,<4.0.0",
"size": 5295,
"upload_time": "2023-07-12T14:17:13",
"upload_time_iso_8601": "2023-07-12T14:17:13.732707Z",
"url": "https://files.pythonhosted.org/packages/a6/d9/323365fd272b489e77d19b38e3045a2be47f0aec785f811f1081dd0097b3/streamz_pulsar-0.1.0.post2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "3705306cb32f021c341fc83738ca3dcb65e841cf389ed5a10e8d0e036a15d434",
"md5": "cc1fccf340f75ba63113216a349eaae3",
"sha256": "5d76476575e98142fe86e1b492d2d618ed42f431f51bdf80d109523b090a9d17"
},
"downloads": -1,
"filename": "streamz_pulsar-0.1.0.post2.tar.gz",
"has_sig": false,
"md5_digest": "cc1fccf340f75ba63113216a349eaae3",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7.2,<4.0.0",
"size": 3607,
"upload_time": "2023-07-12T14:17:15",
"upload_time_iso_8601": "2023-07-12T14:17:15.136684Z",
"url": "https://files.pythonhosted.org/packages/37/05/306cb32f021c341fc83738ca3dcb65e841cf389ed5a10e8d0e036a15d434/streamz_pulsar-0.1.0.post2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-07-12 14:17:15",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "MarekWadinger",
"github_project": "streamz_pulsar",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "streamz-pulsar"
}
JSON
