Stream IoT Sensor Data into Lakehouse Tables with Kafka and Flink CDC

                      
                     
import os
import json
from kafka import KafkaConsumer, KafkaProducer
import mysql.connector
from mysql.connector import Error

# Configuration
KAFKA_SERVER = os.getenv('KAFKA_SERVER', 'localhost:9092')
MYSQL_HOST = os.getenv('MYSQL_HOST', 'localhost')
MYSQL_DATABASE = os.getenv('MYSQL_DATABASE', 'lakehouse')
MYSQL_USER = os.getenv('MYSQL_USER', 'user')
MYSQL_PASSWORD = os.getenv('MYSQL_PASSWORD', 'password')
TOPIC_NAME = 'iot_sensors'

# Initialize MySQL connection
try:
    connection = mysql.connector.connect(
        host=MYSQL_HOST,
        database=MYSQL_DATABASE,
        user=MYSQL_USER,
        password=MYSQL_PASSWORD
    )
    if connection.is_connected():
        print('Connected to MySQL database')
except Error as e:
    print(f'Error connecting to MySQL: {e}')

# Initialize Kafka consumer
consumer = KafkaConsumer(
    TOPIC_NAME,
    bootstrap_servers=KAFKA_SERVER,
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='iot_group'
)

# Function to process and insert data into MySQL
def insert_sensor_data(data: dict):
    try:
        cursor = connection.cursor()
        query = "INSERT INTO sensor_data (sensor_id, value, timestamp) VALUES (%s, %s, %s)"
        cursor.execute(query, (data['sensor_id'], data['value'], data['timestamp']))
        connection.commit()
        print('Data inserted successfully')
    except Error as e:
        print(f'Error inserting data: {e}')
    finally:
        cursor.close()

# Main loop to consume messages
if __name__ == '__main__':
    try:
        for message in consumer:
            sensor_data = json.loads(message.value.decode('utf-8'))
            insert_sensor_data(sensor_data)
    except KeyboardInterrupt:
        print('Shutting down...')
    finally:
        consumer.close()
        if connection.is_connected():
            connection.close()
            print('MySQL connection closed')