Schema validation in producers in Kafka - Mini Project: Build & Apply

Schema validation in producers

📖 Scenario: You are building a Kafka producer that sends user data messages to a topic. To ensure data quality, you want to validate each message against a predefined schema before sending it.

🎯 Goal: Create a Kafka producer that validates messages against a schema before sending them to the Kafka topic.

📋 What You'll Learn

Create a dictionary called user_data with keys "name", "age", and "email" and exact values "Alice", 30, and "[e-mail]"

Create a dictionary called schema that defines the expected types for "name" as str, "age" as int, and "email" as str

Write a function called validate_schema that takes data and schema and returns True if all keys exist and types match, otherwise False

Print "Valid message sent" if the data passes validation, otherwise print "Invalid message, not sent"

💡 Why This Matters

🌍 Real World

Validating data before sending it to Kafka ensures that consumers receive consistent and expected message formats, reducing errors downstream.

💼 Career

Many jobs require working with Kafka producers and ensuring data quality through schema validation to maintain reliable data pipelines.

1

Create the user data dictionary

Create a dictionary called user_data with these exact entries: "name": "Alice", "age": 30, and "email": "[e-mail]"

Kafka

# Create the user_data dictionary with name, age, and email
# Your code here

Need a hint?

Use curly braces {} to create a dictionary with the exact keys and values.

2

Define the schema dictionary

Create a dictionary called schema with keys "name", "age", and "email" and values str, int, and str respectively

Kafka

user_data = {"name": "Alice", "age": 30, "email": "[e-mail]"}
# Create the schema dictionary with expected types
# Your code here

Need a hint?

Use the Python type names str and int as values in the dictionary.

3

Write the schema validation function

Write a function called validate_schema that takes parameters data and schema and returns True if all keys in schema exist in data and their types match, otherwise returns False

Kafka

user_data = {"name": "Alice", "age": 30, "email": "[e-mail]"}
schema = {"name": str, "age": int, "email": str}
# Define the validate_schema function
# Your code here

Need a hint?

Use a for loop to check each key and isinstance() to check the type.

4

Validate and print the result

Use an if statement to call validate_schema(user_data, schema). If it returns True, print "Valid message sent". Otherwise, print "Invalid message, not sent"

Kafka

user_data = {"name": "Alice", "age": 30, "email": "[e-mail]"}
schema = {"name": str, "age": int, "email": str}

def validate_schema(data, schema):
    for key, expected_type in schema.items():
        if key not in data or not isinstance(data[key], expected_type):
            return False
    return True

# Validate user_data and print the result
# Your code here

Need a hint?

Use if to check the function result and print() to show the message.