Introduction to Kafka Stream Processing in Python

Sample output from running our first Faust app

Let us start dissecting the code bit by bit.

The Faust App

app = faust.App('myapp', broker='kafka://localhost:9092')

Here we give our app a name (which will also be used to create a Kafka consumer group) and specify the location of our Kafka brokers. Assuming that you are following this guide with me, you do not need to change the broker settings. Of course, if you have different broker URLs, specify those!

These are the minimum parameters required to get started with Faust. The full list of app parameters can be found here.

We also specify the topic that we want to use using the app.topic() function. We pass in the parameter value_type to force a message structure. This parameter is mostly used for sending messages to a Kafka topic. If we do not need to enforce any message type, we can set value_type to use bytes instead (value_type=bytes).

A message type can be enforced by creating a Python class that extends the class faust.Record.

The Agent

We indicate that a function is a Faust agent by using the @app.agent() decorator. We pass the topic that we want our agent to subscribe to as part of the decorator.

The agent should always be a coroutine. Therefore, we should define the agent function using the async def keywords.

Since our agent is subscribed to a topic, we need to have an asynchronous loop to go over the messages in the stream. We do this using the async for keywords.

Inside this for-loop, we now have access to every single message being consumed by our agent. Here, we can start implementing our stream processing logic.

Besides subscribing an agent to a Kafka topic, we can also have our agent publish messages to some Kafka topic. The agent decorator accepts a parameter — sink — which defines the Kafka topic for publishing messages.

import faust
import logging
from asyncio import sleep
log = logging.getLogger(__name__)
class Test(faust.Record):
    msg: str
app = faust.App('myapp', broker='kafka://localhost:9092')
source_topic = app.topic('test', value_type=Test)
destination_topic = app.topic('test', value_type=Test)
# specify the source_topic and destination_topic to the agent
@app.agent(source_topic, sink=[destination_topic])
async def hello(messages):
    async for message in messages:
        if message is not None:
            log.info(message.msg)
            # the yield keyword is used to send the message to the destination_topic
            yield Test(msg='This message is from the AGENT')
            # sleep for 2 seconds
            await sleep(2)
        else:
            log.info('No message received')
if __name__ == '__main__':
    app.main()

In the above example, we pass in the destination topic as a sink to our agent (in this case, the same test topic we used for our source). On consumption of the first message (which we can send via the producer as shown above), the agent will start executing the logic inside the asynchronous loop.

We are yielding another message to the sink/destination topic. In our case, the destination topic is the same as the source topic, simulating the behavior of constantly receiving messages in our Kafka topic.

Batch Processing

The agent also provides us with the ability to process any Kafka stream in batches. We can achieve this behavior through the stream.take() function.

# specify the source_topic and destination_topic to the agent
@app.agent(source_topic, sink=[destination_topic])
async def hello(messages_stream):
    async for records in messages_stream.take(5, within=10):
        # do something
        yield Test(msg='This message is from the AGENT')

The take() function takes in 2 parameters:

• max_ — the maximum number of messages in the batch
• within — timeout for waiting to receive max_ messages