HomeSoftware EngineeringStreamline Occasion-driven Microservices With Kafka and Python

Streamline Occasion-driven Microservices With Kafka and Python

For a lot of vital utility capabilities, together with streaming and e-commerce, monolithic structure is not ample. With present calls for for real-time occasion information and cloud service utilization, many trendy purposes, akin to Netflix and Lyft, have shifted to an event-driven microservices method. Separated microservices can function independently of each other and improve a code base’s adaptability and scalability.

However what’s an event-driven microservices structure, and why do you have to use it? We’ll study the foundational points and create a whole blueprint for an event-driven microservices venture utilizing Python and Apache Kafka.

Utilizing Occasion-driven Microservices

Occasion-driven microservices mix two trendy structure patterns: microservices architectures and event-driven architectures. Although microservices can pair with request-driven REST architectures, event-driven architectures have gotten more and more related with the rise of huge information and cloud platform environments.

What Is a Microservices Structure?

A microservices structure is a software program growth approach that organizes an utility’s processes as loosely coupled companies. It’s a kind of service-oriented structure (SOA).

In a conventional monolithic construction, all utility processes are inherently interconnected; if one half fails, the system goes down. Microservices architectures as an alternative group utility processes into separate companies interacting with light-weight protocols, offering improved modularity and higher app maintainability and resiliency.

Microservices architecture (with UI individually connected to separate microservices) versus monolithic architecture (with logic and UI connected).
Microservices Structure vs. Monolithic Structure

Although monolithic purposes could also be less complicated to develop, debug, check, and deploy, most enterprise-level purposes flip to microservices as their normal, which permits builders to personal elements independently. Profitable microservices needs to be saved so simple as attainable and talk utilizing messages (occasions) which can be produced and despatched to an occasion stream or consumed from an occasion stream. JSON, Apache Avro, and Google Protocol Buffers are frequent selections for information serialization.

What Is an Occasion-driven Structure?

An event-driven structure is a design sample that buildings software program in order that occasions drive the conduct of an utility. Occasions are significant information generated by actors (i.e., human customers, exterior purposes, or different companies).

Our instance venture options this structure; at its core is an event-streaming platform that manages communication in two methods:

  • Receiving messages from actors that write them (normally known as publishers or producers)
  • Sending messages to different actors that learn them (normally known as subscribers or customers)

In additional technical phrases, our event-streaming platform is software program that acts because the communication layer between companies and permits them to change messages. It might probably implement a wide range of messaging patterns, akin to publish/subscribe or point-to-point messaging, in addition to message queues.

A producer sending a message to an event-streaming platform, which sends the message to one of three consumers.
Occasion-driven Structure

Utilizing an event-driven structure with an event-streaming platform and microservices provides a wealth of advantages:

  • Asynchronous communications: The power to independently multitask permits companies to react to occasions every time they’re prepared as an alternative of ready on a earlier process to complete earlier than beginning the following one. Asynchronous communications facilitate real-time information processing and make purposes extra reactive and maintainable.
  • Full decoupling and adaptability: The separation of producer and shopper elements signifies that companies solely have to work together with the event-streaming platform and the information format they’ll produce or devour. Providers can comply with the single duty precept and scale independently. They will even be carried out by separate growth groups utilizing distinctive expertise stacks.
  • Reliability and scalability: The asynchronous, decoupled nature of event-driven architectures additional amplifies app reliability and scalability (that are already benefits of microservices structure design).

With event-driven architectures, it’s simple to create companies that react to any system occasion. You may also create semi-automatic pipelines that embody some handbook actions. (For instance, a pipeline for automated consumer payouts would possibly embody a handbook safety test triggered by unusually giant payout values earlier than transferring funds.)

Selecting the Venture Tech Stack

We’ll create our venture utilizing Python and Apache Kafka paired with Confluent Cloud. Python is a sturdy, dependable normal for a lot of forms of software program tasks; it boasts a big group and plentiful libraries. It’s a good selection for creating microservices as a result of its frameworks are suited to REST and event-driven purposes (e.g., Flask and Django). Microservices written in Python are additionally generally used with Apache Kafka.

Apache Kafka is a widely known event-streaming platform that makes use of a publish/subscribe messaging sample. It’s a frequent selection for event-driven architectures as a result of its intensive ecosystem, scalability (the results of its fault-tolerance talents), storage system, and stream processing talents.

Lastly, we are going to use Confluent as our cloud platform to effectively handle Kafka and supply out-of-the-box infrastructure. AWS MSK is one other glorious possibility if you happen to’re utilizing AWS infrastructure, however Confluent is simpler to arrange as Kafka is the core a part of its system and it provides a free tier.

Implementing the Venture Blueprint

We’ll arrange our Kafka microservices instance in Confluent Cloud, create a easy message producer, then arrange and enhance it to optimize scalability. By the tip of this tutorial, we may have a functioning message producer that efficiently sends information to our cloud cluster.

Kafka Setup

We’ll first create a Kafka cluster. Kafka clusters host Kafka servers that facilitate communication. Producers and customers interface with the servers utilizing Kafka matters (classes storing information).

  1. Join Confluent Cloud. When you create an account, the welcome web page seems with choices for creating a brand new Kafka cluster. Choose the Primary configuration.
  2. Select a cloud supplier and area. It is best to optimize your selections for the most effective cloud ping outcomes out of your location. One possibility is to decide on AWS and carry out a cloud ping check (click on HTTP Ping) to establish the most effective area. (For the scope of our tutorial, we are going to depart the “Single zone” possibility chosen within the “Availability” discipline.)
  3. The subsequent display screen asks for a cost setup, which we are able to skip since we’re on a free tier. After that, we are going to enter our cluster identify (e.g., “MyFirstKafkaCluster”), affirm our settings, and choose Launch cluster.
The Confluent “Create cluster” screen with various configuration choices for the “MyFirstKafkaCluster” cluster and a “Launch cluster” button.
Kafka Cluster Configuration

With a working cluster, we’re able to create our first matter. Within the left-hand menu bar, navigate to Subjects and click on Create matter. Add a subject identify (e.g., “MyFirstKafkaTopic”) and proceed with the default configurations (together with setting six partitions).

Earlier than creating our first message, we should arrange our shopper. We will simply Configure a shopper from our newly created matter overview (alternatively, within the left-hand menu bar, navigate to Shoppers). We’ll use Python as our language after which click on Create Kafka cluster API key.

The Confluent Clients screen showing step 2 (client code configuration) with the Kafka cluster API key setup and the configuration code snippet.
Kafka Cluster API Key Setup

At this level, our event-streaming platform is lastly able to obtain messages from our producer.

Easy Message Producer

Our producer generates occasions and sends them to Kafka. Let’s write some code to create a easy message producer. I like to recommend organising a digital atmosphere for our venture since we might be putting in a number of packages in our surroundings.

First, we are going to add our surroundings variables from the API configuration from Confluent Cloud. To do that in our digital atmosphere, we’ll add export SETTING=worth for every setting under to the tip of our activate file (alternatively, you’ll be able to add SETTING=worth to your .env file):

export KAFKA_BOOTSTRAP_SERVERS=<bootstrap.servers>
export KAFKA_SECURITY_PROTOCOL=<safety.protocol>
export KAFKA_SASL_MECHANISMS=<sasl.mechanisms>
export KAFKA_SASL_USERNAME=<sasl.username>
export KAFKA_SASL_PASSWORD=<sasl.password>

Ensure that to exchange every entry together with your Confluent Cloud values (for instance, <sasl.mechanisms> needs to be PLAIN), together with your API key and secret because the username and password. Run supply env/bin/activate, then printenv. Our new settings ought to seem, confirming that our variables have been accurately up to date.

We might be utilizing two Python packages:

We’ll run the command pip set up confluent-kafka python-dotenv to put in these. There are a lot of different packages for Kafka in Python that could be helpful as you broaden your venture.

Lastly, we’ll create our primary producer utilizing our Kafka settings. Add a file:

import os

from confluent_kafka import KafkaException, Producer
from dotenv import load_dotenv

def principal():
    settings = {
        'bootstrap.servers': os.getenv('KAFKA_BOOTSTRAP_SERVERS'),
        'safety.protocol': os.getenv('KAFKA_SECURITY_PROTOCOL'),
        'sasl.mechanisms': os.getenv('KAFKA_SASL_MECHANISMS'),
        'sasl.username': os.getenv('KAFKA_SASL_USERNAME'),
        'sasl.password': os.getenv('KAFKA_SASL_PASSWORD'),

    producer = Producer(settings)
    producer.flush()  # Look forward to the affirmation that the message was acquired

if __name__ == '__main__':

With this easy code we create our producer and ship it a easy check message. To check the outcome, run python3

Confluent’s Cluster Overview dashboard, with one spike appearing in the Production (bytes/sec) and Storage graphs, and no data shown for Consumption.
First Take a look at Message Throughput and Storage

Checking our Kafka cluster’s Cluster Overview > Dashboard, we are going to see a brand new information level on our Manufacturing graph for the message despatched.

Customized Message Producer

Our producer is up and operating. Let’s reorganize our code to make our venture extra modular and OOP-friendly. This can make it simpler so as to add companies and scale our venture sooner or later. We’ll break up our code into 4 information:

  • Holds our Kafka configurations.
  • Comprises a customized produce() methodology and error dealing with.
  • Handles completely different enter information varieties.
  • Runs our closing app utilizing our customized lessons.

First, our KafkaSettings class will encapsulate our Apache Kafka settings, so we are able to simply entry these from our different information with out repeating code:

import os

class KafkaSettings:
    def __init__(self):
                      self.conf = {
            'bootstrap.servers': os.getenv('KAFKA_BOOTSTRAP_SERVERS'),
            'safety.protocol': os.getenv('KAFKA_SECURITY_PROTOCOL'),
            'sasl.mechanisms': os.getenv('KAFKA_SASL_MECHANISMS'),
            'sasl.username': os.getenv('KAFKA_SASL_USERNAME'),
            'sasl.password': os.getenv('KAFKA_SASL_PASSWORD'),

Subsequent, our KafkaProducer permits us to customise our produce() methodology with help for varied errors (e.g., an error when the message dimension is simply too giant), and in addition routinely flushes messages as soon as produced:

from confluent_kafka import KafkaError, KafkaException, Producer

from kafka_producer_message import ProducerMessage
from kafka_settings import KafkaSettings

class KafkaProducer:
    def __init__(self, settings: KafkaSettings):
        self._producer = Producer(settings.conf)

    def produce(self, message: ProducerMessage):
            self._producer.produce(message.matter, key=message.key, worth=message.worth)
        besides KafkaException as exc:
            if exc.args[0].code() == KafkaError.MSG_SIZE_TOO_LARGE:
                move  # Deal with the error right here
                increase exc

In our instance’s try-except block, we skip over the message whether it is too giant for the Kafka cluster to devour. Nevertheless, it is best to replace your code in manufacturing to deal with this error appropriately. Consult with the confluent-kafka documentation for an entire checklist of error codes.

Now, our ProducerMessage class handles several types of enter information and accurately serializes them. We’ll add performance for dictionaries, Unicode strings, and byte strings:

import json

class ProducerMessage:
    def __init__(self, matter: str, worth, key=None) -> None:
        self.matter = f'{matter}'
        self.key = key
        self.worth = self.convert_value_to_bytes(worth)

    def convert_value_to_bytes(cls, worth):
        if isinstance(worth, dict):
            return cls.from_json(worth)

        if isinstance(worth, str):
            return cls.from_string(worth)

        if isinstance(worth, bytes):
            return cls.from_bytes(worth)

        increase ValueError(f'Fallacious message worth kind: {kind(worth)}')

    def from_json(cls, worth):
        return json.dumps(worth, indent=None, sort_keys=True, default=str, ensure_ascii=False)

    def from_string(cls, worth):
        return worth.encode('utf-8')

    def from_bytes(cls, worth):
        return worth

Lastly, we are able to construct our app utilizing our newly created lessons in

from dotenv import load_dotenv

from kafka_producer import KafkaProducer
from kafka_producer_message import ProducerMessage
from kafka_settings import KafkaSettings

def principal():
    settings = KafkaSettings()
    producer = KafkaProducer(settings)
    message = ProducerMessage(
        worth={"worth": "MyFirstKafkaValue"},

if __name__ == '__main__':

We now have a neat abstraction above the confluent-kafka library. Our customized producer possesses the identical performance as our easy producer with added scalability and adaptability, able to adapt to numerous wants. We might even change the underlying library totally if we wished to, which units our venture up for fulfillment and long-term maintainability.

Confluent’s Cluster Overview dashboard: Production shows two spikes, Storage shows two steps (with horizontal lines), and Consumption shows no data.
Second Take a look at Message Throughput and Storage

After operating python3, we see but once more that information has been despatched to our cluster within the Cluster Overview > Dashboard panel of Confluent Cloud. Having despatched one message with the straightforward producer, and a second with our customized producer, we now see two spikes in manufacturing throughput and a rise in general storage used.

Trying Forward: From Producers to Customers

An event-driven microservices structure will improve your venture and enhance its scalability, flexibility, reliability, and asynchronous communications. This tutorial has given you a glimpse of those advantages in motion. With our enterprise-scale producer up and operating, sending messages efficiently to our Kafka dealer, the following steps could be to create a shopper to learn these messages from different companies and add Docker to our utility.

The editorial workforce of the Toptal Engineering Weblog extends its gratitude to E. Deniz Toktay for reviewing the code samples and different technical content material introduced on this article.

Additional Studying on the Toptal Engineering Weblog:


Most Popular

Recent Comments