Streamline Occasion-driven Microservices With Kafka and Python

For a lot of crucial utility capabilities, together with streaming and e-commerce, monolithic structure is now not adequate. With present calls for for real-time occasion information and cloud service utilization, many trendy functions, comparable to Netflix and Lyft, have shifted to an event-driven microservices strategy. 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 must you use it? We’ll look at the foundational features and create a whole blueprint for an event-driven microservices challenge 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 massive information and cloud platform environments.

What Is a Microservices Structure?

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

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

Although monolithic functions could also be easier to develop, debug, check, and deploy, most enterprise-level functions flip to microservices as their customary, which permits builders to personal parts independently. Profitable microservices must be stored so simple as attainable and talk utilizing messages (occasions) which are produced and despatched to an occasion stream or consumed from an occasion stream. JSON, Apache Avro, and Google Protocol Buffers are widespread decisions for information serialization.

What Is an Occasion-driven Structure?

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

Our instance challenge 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 providers and permits them to trade messages. It may possibly implement quite a lot of messaging patterns, comparable to publish/subscribe or point-to-point messaging, in addition to message queues.

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

• Asynchronous communications: The power to independently multitask permits providers to react to occasions each time they’re prepared as a substitute of ready on a earlier process to complete earlier than beginning the following one. Asynchronous communications facilitate real-time information processing and make functions extra reactive and maintainable.
• Full decoupling and suppleness: The separation of producer and client parts signifies that providers solely must work together with the event-streaming platform and the info format they will produce or devour. Providers can comply with the single duty precept and scale independently. They will even be carried out by separate improvement 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 straightforward to create providers that react to any system occasion. It’s also possible to create semi-automatic pipelines that embody some guide actions. (For instance, a pipeline for automated person payouts may embody a guide safety examine triggered by unusually giant payout values earlier than transferring funds.)

Selecting the Challenge Tech Stack

We are going to create our challenge utilizing Python and Apache Kafka paired with Confluent Cloud. Python is a sturdy, dependable customary for a lot of sorts of software program initiatives; 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 functions (e.g., Flask and Django). Microservices written in Python are additionally generally used with Apache Kafka.

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

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 wonderful choice for those who’re utilizing AWS infrastructure, however Confluent is less complicated to arrange as Kafka is the core a part of its system and it gives a free tier.

Implementing the Challenge Blueprint

We’ll arrange our Kafka microservices instance in Confluent Cloud, create a easy message producer, then manage 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 subjects (classes storing data).

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. You need to optimize your decisions for one of the best cloud ping outcomes out of your location. One choice is to decide on AWS and carry out a cloud ping check (click on HTTP Ping) to determine one of the best area. (For the scope of our tutorial, we are going to depart the “Single zone” choice chosen within the “Availability” discipline.)
3. The following display asks for a fee setup, which we are able to skip since we’re on a free tier. After that, we are going to enter our cluster title (e.g., “MyFirstKafkaCluster”), affirm our settings, and choose Launch cluster.

With a working cluster, we’re able to create our first matter. Within the left-hand menu bar, navigate to Matters and click on Create matter. Add a subject title (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 Purchasers). We’ll use Python as our language after which click on Create Kafka cluster API key.

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 establishing a digital setting for our challenge since we can be putting in a number of packages in the environment.

First, we are going to add the environment variables from the API configuration from Confluent Cloud. To do that in our digital setting, we’ll add export SETTING=worth for every setting beneath to the tip of our activate file (alternatively, you possibly can 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>

Be certain to interchange every entry along with your Confluent Cloud values (for instance, <sasl.mechanisms> must be PLAIN), along 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 can be utilizing two Python packages:

We’ll run the command pip set up confluent-kafka python-dotenv to put in these. There are various different packages for Kafka in Python which may be helpful as you broaden your challenge.

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

# simple_producer.py
import os

from confluent_kafka import KafkaException, Producer
from dotenv import load_dotenv

def important():
    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.produce(
        matter='MyFirstKafkaTopic',
                      key=None,
                      worth='MyFirstValue-111',
    )
    producer.flush()  # Watch for the affirmation that the message was obtained

if __name__ == '__main__':
    load_dotenv()
    important()

With this easy code we create our producer and ship it a easy check message. To check the end result, run python3 simple_producer.py:

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 working. Let’s reorganize our code to make our challenge extra modular and OOP-friendly. This may make it simpler so as to add providers and scale our challenge sooner or later. We’ll break up our code into 4 information:

• kafka_settings.py: Holds our Kafka configurations.
• kafka_producer.py: Incorporates a customized produce() methodology and error dealing with.
• kafka_producer_message.py: Handles totally different enter information varieties.
• advanced_producer.py: Runs our closing app utilizing our customized courses.

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:

# kafka_settings.py
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 assist for numerous errors (e.g., an error when the message dimension is just too giant), and in addition robotically flushes messages as soon as produced:

# kafka_producer.py
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):
        strive:
            self._producer.produce(message.matter, key=message.key, worth=message.worth)
            self._producer.flush()
        besides KafkaException as exc:
            if exc.args[0].code() == KafkaError.MSG_SIZE_TOO_LARGE:
                cross  # Deal with the error right here
            else:
                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. Nonetheless, it is best to replace your code in manufacturing to deal with this error appropriately. Confer with the confluent-kafka documentation for an entire record 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:

# kafka_producer_message.py
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)

    @classmethod
    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'Unsuitable message worth kind: {kind(worth)}')

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

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

    @classmethod
    def from_bytes(cls, worth):
        return worth

Lastly, we are able to construct our app utilizing our newly created courses in advanced_producer.py:

# advanced_producer.py
from dotenv import load_dotenv

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

def important():
    settings = KafkaSettings()
    producer = KafkaProducer(settings)
    message = ProducerMessage(
        matter='MyFirstKafkaTopic',
        worth={"worth": "MyFirstKafkaValue"},
        key=None,
    )
    producer.produce(message)

if __name__ == '__main__':
    load_dotenv()
    important()

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 suppleness, able to adapt to varied wants. We might even change the underlying library totally if we wished to, which units our challenge up for achievement and long-term maintainability.

After working python3 advanced_producer.py, 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 Shoppers

An event-driven microservices structure will improve your challenge 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 working, sending messages efficiently to our Kafka dealer, the following steps could be to create a client to learn these messages from different providers and add Docker to our utility.

The editorial crew 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: