Using a Partition Key#

Partition keys are a crucial concept in Apache Kafka, enabling you to determine the appropriate partition for a message. This ensures that related messages are kept together in the same partition, which can be invaluable for maintaining order or grouping related messages for efficient processing. Additionally, Kafka utilizes partitioning to distribute load across multiple brokers and scale horizontally, while replicating data across brokers provides fault tolerance.

You can specify your partition key when calling publish(...) on a publisher created with broker.publisher(...) in FastStream. This guide will walk you through the process of using partition keys effectively.

Publishing with a Partition Key#

To publish a message to a Kafka topic using a partition key, follow these steps:

Step 1: Define the Publisher#

In your FastStream application, define the publisher using broker.publisher(...). It accepts a default key (and partition) for every message it sends; a key passed to publish() overrides that default. This example sets no default and passes the key with each publish call instead.

1

to_output_data = broker.publisher("output_data")

Step 2: Pass the Key#

When you're ready to publish a message with a specific key, simply include the key parameter in the publish function call. This key parameter is used to determine the appropriate partition for the message.

1

await to_output_data.publish(Data(data=msg.data + 1.0), key=b"key")

Example Application#

Let's examine a complete application example that consumes messages from the "input_data" topic and publishes them with a specified key to the "output_data" topic. This example will illustrate how to incorporate partition keys into your Kafka-based applications:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from pydantic import BaseModel, Field, NonNegativeFloat

from faststream import Context, FastStream, Logger
from faststream.confluent import KafkaBroker


class Data(BaseModel):
    data: NonNegativeFloat = Field(
        ..., examples=[0.5], description="Float data example",
    )


broker = KafkaBroker("localhost:9092")
app = FastStream(broker)


to_output_data = broker.publisher("output_data")


@broker.subscriber("input_data")
async def on_input_data(
    msg: Data, logger: Logger, key: bytes = Context("message.raw_message.key"),
) -> None:
    logger.info("on_input_data(msg=%s)", msg)
    await to_output_data.publish(Data(data=msg.data + 1.0), key=b"key")

As you can see, the primary difference from standard publishing is the inclusion of the key parameter in the publish call. This key parameter is essential for controlling how Kafka partitions and processes your messages.

In summary, using partition keys in Apache Kafka is a fundamental practice for optimizing message distribution, maintaining order, and achieving efficient processing. It is a key technique for ensuring that your Kafka-based applications scale gracefully and handle large volumes of data effectively.