Faust-streaming is a Python stream processing library that ports concepts from Kafka Streams to Python. It enables building high-performance distributed systems and real-time data pipelines. This project is an actively maintained fork of the original Faust library, aiming for continuous releases, improved code quality, and support for the latest Kafka drivers. The current version is 0.11.3, with releases happening periodically based on community contributions and dependency updates.
pip install faust-streamingVerified import paths — ran on the pinned version, not inferred.
This quickstart demonstrates a basic Faust streaming application. It initializes a Faust application, defines a `Record` model for incoming data, sets up a Kafka topic, and creates an `@app.agent` to consume and process messages from that topic. The Kafka broker address is configurable via an environment variable.
Ensure `aiokafka` is updated to a compatible version (>=0.10.0, preferably latest stable) when upgrading `faust-streaming`. Review any custom code interacting directly with `aiokafka` internals.
Monitor `stream_processing_timeout` and agent logs for errors. Consider increasing `stream_processing_timeout` if processing individual events takes longer than the default. Review agent logic for potential deadlocks or long-running synchronous operations.
Ensure `mode-streaming` is installed at version `0.4.0` or higher to avoid import errors and ensure correct functionality when using `faust-streaming >=0.11.0`.
Plan for schema evolution. For topic type changes, implement an upgrade path (e.g., new topic for new schema) and ensure all consumers/producers are updated. For model renames, ensure old model names are still resolvable or provide a migration.
Design agents that modify tables to run with a concurrency of 1, or ensure that only non-concurrent agents perform table modifications. Concurrent agents should only read from tables.
Pin your aiokafka dependency to a compatible older version (e.g., `pip install "aiokafka<0.11.0"`) or upgrade faust-streaming to its latest version which might have updated aiokafka compatibility.
Ensure the `faust -A your_app_module worker` command is executed from a directory where `your_app_module.py` (or the package containing it) is importable. For example, if your app is in `myproject/app.py`, run `faust -A myproject.app worker` from the directory containing `myproject`.
Verify that the Kafka broker address and port specified in `faust.App(..., broker='kafka://...')` are correct. Ensure the Kafka broker is running and is reachable from the machine where the Faust worker is being started. Check network configurations and firewall settings.
Implement error handling around `stream.current_event` access, or ensure that you are only accessing `current_event` inside the `async for` loop where an event is guaranteed to be present. Consider the `stream_wait_empty` setting in your `faust.App` configuration.
Ensure that when using Avro models, all named types (including enums and nested records within unions) have unique names or are properly namespaced. For `faust.Record` classes, you may need to define a `Meta` class with a `namespace` attribute.