A Python implementation of the state stores concept within the Kafka Streams API
Gregor Samsa was the unfortunate soul in the Kafka classic 'The Metamorphosis (1915)' who woke up one day as a giant insect.
This uses SQLite by default but also supports RocksDB by specifying store='rocksdb'
in the PersistentConsumer init.
brew install cmake
if you haven't got it already
git clone https://github.com/twmht/python-rocksdb.git --recursive -b pybind11
python setup.py install
inside the python-rocksdb directory.
You can run rocksdb locally on OSX with brew install rocksdb
. It's not a DB in the traditional sense, but really just
a super fast way to store stuff on disk (as fast as that really is!).
The following is a pretty simple way of pulling data from kafka, and saving it to a local store. The default is sqlite but rocksdb is also possible. The example below:
- Initialises a consumer with a table called 'node_status'
- Queries the local store for the received key.
- Compares the received value with a saved value in the local store.
- It then saves the message to the store for future computation.
from samsa import PersistentConsumer
def loop():
topic = "nodes.status"
with PersistentConsumer(
[topic], table_name="node_status", group_id=KAFKA_GROUP_ID,
bootstrap_servers="localhost:9092"
) as store:
while True:
msg = store.consume(process_callback)
if msg:
key = msg.key().decode("utf-8")
value = json.loads(msg.value().decode("utf-8"))["data"]["status"]
status = store.query(key)
if status != value:
print(f"Status has changed from {status} to {value} for {key}!")
else:
print("No status change, nothing to see here!")
store.save(key, value)
if __name__ == "__main__":
loop()