Transform and converter chains in Kafka - Mini Project: Build & Apply

Transform and Converter Chains in Kafka

📖 Scenario: You are working with Kafka Connect to process streaming data. You want to transform the data records by chaining multiple transformations and converters to prepare the data for your target system.

🎯 Goal: Build a Kafka Connect configuration that chains a ReplaceField transformation and a ValueToKey transformation, then uses JSON converters for both key and value.

📋 What You'll Learn

Create a Kafka Connect configuration dictionary named connector_config with the specified settings

Add a transforms property listing the transformations in order: replaceField and valueToKey

Configure the ReplaceField transformation to keep only the id and name fields

Configure the ValueToKey transformation to use the id field as the key

Set the key and value converters to org.apache.kafka.connect.json.JsonConverter with schemas disabled

Print the final connector_config dictionary

💡 Why This Matters

🌍 Real World

Kafka Connect is used to move data between Kafka and other systems. Transform and converter chains help prepare data by changing its shape and format before sending it to the destination.

💼 Career

Understanding how to configure transform and converter chains is important for data engineers and developers working with Kafka to ensure data quality and compatibility.

1

Create the initial Kafka Connect configuration dictionary

Create a dictionary called connector_config with these exact entries: "name": "my-connector", "connector.class": "FileStreamSource", and "tasks.max": "1".

Kafka

# Create the connector_config dictionary with name, connector.class, and tasks.max
# Your code here

Need a hint?

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

2

Add the transforms property with the transform chain

Add to connector_config a key "transforms" with the value "replaceField,valueToKey" to specify the transform chain order.

Kafka

connector_config = {
    "name": "my-connector",
    "connector.class": "FileStreamSource",
    "tasks.max": "1"
}
# Add the transforms chain to connector_config
# Your code here

Need a hint?

Add the "transforms" key with the exact string value listing the transforms separated by commas.

3

Configure the ReplaceField and ValueToKey transformations and converters

Add these exact entries to connector_config: "transforms.replaceField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value", "transforms.replaceField.whitelist": "id,name", "transforms.valueToKey.type": "org.apache.kafka.connect.transforms.ValueToKey", "transforms.valueToKey.fields": "id", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "key.converter.schemas.enable": "false", "value.converter": "org.apache.kafka.connect.json.JsonConverter", and "value.converter.schemas.enable": "false".

Kafka

connector_config = {
    "name": "my-connector",
    "connector.class": "FileStreamSource",
    "tasks.max": "1",
    "transforms": "replaceField,valueToKey"
}
# Add transform types, fields, and converters to connector_config
# Your code here

Need a hint?

Each transform and converter setting must be added as a separate key-value pair in the dictionary.

4

Print the final connector_config dictionary

Write print(connector_config) to display the complete Kafka Connect configuration dictionary.

Kafka

connector_config = {
    "name": "my-connector",
    "connector.class": "FileStreamSource",
    "tasks.max": "1",
    "transforms": "replaceField,valueToKey",
    "transforms.replaceField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.replaceField.whitelist": "id,name",
    "transforms.valueToKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
    "transforms.valueToKey.fields": "id",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
}
# Print the connector_config dictionary
# Your code here

Need a hint?

Use the print() function to show the dictionary.