Faust Streaming

Warnings

• breaking Upgrading to v0.11.2 changed the internal `aiokafka` topic error handling (`topic_errors` was renamed) and bumped the minimum `aiokafka` version to `0.10.0`. Older `aiokafka` versions may cause compatibility issues.
  Fix: 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.
• gotcha Users reported that consumers could 'slowly die over time' or stop receiving messages after upgrading to `v0.11.2` from `v0.11.1`. This might be related to stream processing timeouts or agent hanging.
  Fix: 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.
• breaking Version `0.11.0` included fixes for imports from `mode-streaming~=0.4.0`. This indicates a strong compatibility requirement with `mode-streaming` version `0.4.0` or newer.
  Fix: 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`.
• breaking Changing the `key_type` or `value_type` of an existing topic is a backward-incompatible change. All Faust instances using the old types must be restarted. Renaming model classes can also cause deserialization errors if not handled with an upgrade strategy.
  Fix: 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.
• gotcha Concurrent agents are explicitly not allowed to modify tables. Attempting to do so will raise an exception.
  Fix: 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.

Install

• pip install faust-streaming Install stable version

Imports

• App

  from faust import App

• Record

  from faust import Record

Quickstart

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.

import faust
import os

# Configure Kafka broker from environment variable or default to localhost
KAFKA_BROKER = os.environ.get('FAUST_BROKER', 'kafka://localhost:9092')

app = faust.App(
    'my-streaming-app',
    broker=KAFKA_BROKER,
    value_serializer='json',
)

# Define a data model for messages
class Order(faust.Record):
    account_id: str
    amount: float
    item_id: str

# Define an input topic using the Order model
orders_topic = app.topic('orders', value_type=Order)

# Define an agent to process messages from the 'orders' topic
@app.agent(orders_topic)
async def process_orders(orders):
    async for order in orders:
        print(f"Processing order {order.item_id} for account {order.account_id}: ${order.amount}")
        # Example: Perform some asynchronous operation or send to another topic
        # await another_topic.send(value={'processed': True, 'order_id': order.item_id})

# To run this worker: save as app.py, then execute 'faust -A app worker -l info'