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

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

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

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

Hint

Use the print() function to show the dictionary.

Practice

1. What is the main purpose of using transform and converter chains in Kafka Connect?

easy

A. To monitor Kafka cluster health

B. To store data permanently in Kafka topics

C. To create new Kafka topics automatically

D. To change data step-by-step during processing

Solution

1. Step 1: Understand transform and converter chains

   Transform and converter chains let you modify data in stages as it moves through Kafka Connect.

2. Step 2: Identify the main purpose

   The main goal is to change or clean data step-by-step automatically before it reaches its destination.

3. Final Answer:

   To change data step-by-step during processing -> Option D

4. Quick Check:

   Transform chains = step-by-step data change [OK]

Hint: Transforms modify data stepwise in Kafka Connect [OK]

Common Mistakes:

• Confusing transform chains with topic creation
• Thinking transforms monitor cluster health
• Assuming transforms store data permanently

2. Which of the following is the correct way to list multiple transforms in Kafka Connect configuration?

easy

A. transforms=InsertField,ReplaceField

B. transforms=[InsertField, ReplaceField]

C. transforms: InsertField; ReplaceField

D. transforms = InsertField ReplaceField

Solution

1. Step 1: Recall Kafka Connect transform list syntax

   Transforms are listed as a comma-separated string without brackets or semicolons.

2. Step 2: Match the correct syntax

   transforms=InsertField,ReplaceField uses commas correctly: transforms=InsertField,ReplaceField.

3. Final Answer:

   transforms=InsertField,ReplaceField -> Option A

4. Quick Check:

   Comma-separated list = correct syntax [OK]

Hint: Use commas to separate transforms in config [OK]

Common Mistakes:

• Using brackets or semicolons instead of commas
• Adding spaces around commas incorrectly
• Using colons or spaces without commas

3. Given this Kafka Connect transform chain configuration:
transforms=InsertField,ReplaceField
transforms.InsertField.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.InsertField.static.field=source
transforms.InsertField.static.value=app1
transforms.ReplaceField.type=org.apache.kafka.connect.transforms.ReplaceField$Value
transforms.ReplaceField.renames=oldName:newName
What will happen to a record with value {"oldName":"data"}?

medium

A. The record value becomes {"newName":"data"} only

B. The record value becomes {"oldName":"data", "source":"app1"}

C. The record value becomes {"source":"app1", "newName":"data"}

D. The record value remains unchanged

Solution

1. Step 1: Apply InsertField transform

   This adds a field "source" with value "app1" to the record, so value becomes {"oldName":"data", "source":"app1"}.

2. Step 2: Apply ReplaceField transform

   This renames "oldName" to "newName", so final value is {"source":"app1", "newName":"data"}.

3. Final Answer:

   {"source":"app1", "newName":"data"} -> Option C

4. Quick Check:

   InsertField adds, ReplaceField renames = final value [OK]

Hint: Apply transforms in order listed [OK]

Common Mistakes:

• Applying transforms in wrong order
• Forgetting InsertField adds a new field
• Assuming ReplaceField deletes old field without rename

4. You configured a transform chain but get an error: "Unknown transform type". Which of these fixes the problem?

medium

A. Remove all transforms from the config

B. Correct the transform type class name in the config

C. Restart Kafka Connect without changing config

D. Increase Kafka Connect heap size

Solution

1. Step 1: Understand the error cause

   "Unknown transform type" means Kafka Connect cannot find the class name specified for the transform.

2. Step 2: Fix the transform type

   Correcting the transform type class name in the configuration ensures Kafka Connect can load it properly.

3. Final Answer:

   Correct the transform type class name in the config -> Option B

4. Quick Check:

   Fix class name to resolve unknown transform error [OK]

Hint: Check transform class names carefully [OK]

Common Mistakes:

• Ignoring typo in transform type
• Restarting without fixing config
• Assuming memory issues cause this error

5. You want to create a transform chain that first removes sensitive fields, then adds a timestamp, and finally converts the record to JSON string. Which order of transforms is correct?

hard

A. RemoveFields, InsertField (timestamp), StringConverter

B. StringConverter, RemoveFields, InsertField (timestamp)

C. InsertField (timestamp), RemoveFields, StringConverter

D. RemoveFields, StringConverter, InsertField (timestamp)

Solution

1. Step 1: Remove sensitive fields first

   Removing fields must happen before adding new data or converting formats to avoid leaking sensitive info.

2. Step 2: Add timestamp after cleaning

   InsertField adds the timestamp field next, so the record has the new info before conversion.

3. Step 3: Convert record to JSON string last

   StringConverter should be last to convert the final cleaned and enriched record into a JSON string.

4. Final Answer:

   RemoveFields, InsertField (timestamp), StringConverter -> Option A

5. Quick Check:

   Clean, enrich, then convert = correct order [OK]

Hint: Clean data first, then add fields, convert last [OK]

Common Mistakes:

• Converting to string before removing fields
• Adding fields after conversion
• Wrong transform order causing data errors