Source connectors in Kafka - Mini Project: Build & Apply

Kafka Source Connector Setup

📖 Scenario: You are working on a data pipeline that collects data from a database and sends it to Kafka topics for processing.

🎯 Goal: Set up a Kafka source connector configuration to pull data from a MySQL database into Kafka.

📋 What You'll Learn

Create a JSON configuration for a Kafka source connector

Specify the connector class for MySQL

Set the connection URL, user, and password

Define the Kafka topic prefix

Print the final JSON configuration

💡 Why This Matters

🌍 Real World

Kafka source connectors are used to automatically pull data from databases or other systems into Kafka topics for real-time processing.

💼 Career

Understanding how to configure Kafka connectors is important for data engineers and developers working with streaming data pipelines.

1

Create the base JSON configuration

Create a variable called connector_config and assign it a dictionary with the key "name" set to "mysql-source-connector".

Kafka

# Create the base connector_config dictionary
# Your code here

Need a hint?

Use a dictionary with the key 'name' and value 'mysql-source-connector'.

2

Add connector class and database connection details

Add to connector_config a key "config" with a nested dictionary containing these exact entries: "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "connection.url": "jdbc:mysql://localhost:3306/mydb", "connection.user": "user", and "connection.password": "password".

Kafka

connector_config = {"name": "mysql-source-connector"}
# Add the config dictionary with connector class and DB details
# Your code here

Need a hint?

Add a 'config' key with a nested dictionary containing the connector class and connection details.

3

Add topic prefix and table whitelist

Inside connector_config["config"], add the keys "topic.prefix" with value "mysql-" and "table.whitelist" with value "customers".

Kafka

connector_config = {"name": "mysql-source-connector", "config": {"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "connection.url": "jdbc:mysql://localhost:3306/mydb", "connection.user": "user", "connection.password": "password"}}
# Add topic.prefix and table.whitelist to the config dictionary
# Your code here

Need a hint?

Add 'topic.prefix' and 'table.whitelist' keys inside the 'config' dictionary.

4

Print the connector configuration JSON

Import the json module and print the JSON string of connector_config using json.dumps() with indentation of 2.

Kafka

connector_config = {"name": "mysql-source-connector", "config": {"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "connection.url": "jdbc:mysql://localhost:3306/mydb", "connection.user": "user", "connection.password": "password", "topic.prefix": "mysql-", "table.whitelist": "customers"}}
# Import json and print the JSON string of connector_config with indent=2
# Your code here

Need a hint?

Use json.dumps(connector_config, indent=2) inside print to display the JSON nicely.