kafka数据入mysql库

将Kafka数据入库到MySQL有多种方法，以下是其中一种常用的方法：

1. 创建一个Kafka消费者，以获取Kafka中的数据。

2. 将消费者获取到的数据转换成MySQL的数据格式。

3. 将转换后的MySQL数据插入到MySQL数据库中。

具体实现步骤如下：

1. 创建一个Kafka消费者，获取Kafka中的数据。

from kafka import KafkaConsumer

consumer = KafkaConsumer('test-topic', bootstrap_servers=['localhost:9092'])
for message in consumer:
    print(message.value)

2. 将消费者获取到的数据转换成MySQL的数据格式。这里假设Kafka中的数据是JSON格式，需要将其转换成MySQL格式。

import json

def json_to_mysql(json_data):
    mysql_data = {}
    mysql_data['id'] = json_data['id']
    mysql_data['name'] = json_data['name']
    mysql_data['age'] = json_data['age']
    return mysql_data

json_data = '{"id": 1, "name": "Alice", "age": 20}'
mysql_data = json_to_mysql(json.loads(json_data))
print(mysql_data)

输出结果：

{'id': 1, 'name': 'Alice', 'age': 20}

3. 将转换后的MySQL数据插入到MySQL数据库中。

import pymysql

db = pymysql.connect('localhost', 'root', 'password', 'testdb')
cursor = db.cursor()

sql = "INSERT INTO students(id, name, age) VALUES (%s, %s, %s)"
val = (mysql_data['id'], mysql_data['name'], mysql_data['age'])
cursor.execute(sql, val)

db.commit()
db.close()

以上是一个简单的将Kafka数据入MySQL库的实现方法，具体的实现会根据实际情况有所不同。