python3实现从kafka获取数据,并解析为json格式,写入到mysql中

在这里我将为您提供一个完整的攻略，讲解如何用Python3从Kafka中获取数据，并将其解析为JSON格式，在将数据写入MySQL中的过程。

准备工作

在开始之前，需要先确保以下环境已经安装：

• Python3: 用于编写和执行Python代码
• pip: 用于安装Python第三方包
• kafka-python: 用于连接到Kafka并获取数据
• pymysql: 用于连接到MySQL并执行数据库操作

可以通过以下命令安装：

pip install kafka-python
pip install pymysql

连接Kafka并获取数据

以下是从Kafka中获取数据的示例代码。在这个示例中，我们将从名为“KafkaTest”的Kafka主题中获取数据，并将其发送到控制台。

from kafka import KafkaConsumer
import json

# 连接到Kafka服务器
consumer = KafkaConsumer('KafkaTest',
                         bootstrap_servers=['localhost:9092'],
                         auto_offset_reset='earliest',                                                  
                         enable_auto_commit=True,
                         group_id='json-group')

# 读取Kafka中的数据
for message in consumer:
    # 将数据解析为JSON格式
    data = json.loads(message.value)

    # 将JSON数据打印到控制台
    print(data)

在这个示例中，我们首先通过调用KafkaConsumer函数来连接到Kafka服务器，并指定了主题名称、服务器地址和组ID等参数。然后，我们使用for循环来遍历从Kafka中获取的每一条消息，并将其解析为JSON格式后输出到控制台。

连接MySQL并将数据写入

以下是将数据写入MySQL数据库的示例代码。在这个示例中，我们将连接到名为“MyDB”的数据库，并将数据写入名为“KafkaData”的表中。

import pymysql.cursors

# 连接MySQL数据库
connection = pymysql.connect(host='localhost',
                             user='root',
                             password='password',
                             db='MyDB',
                             charset='utf8mb4',
                             cursorclass=pymysql.cursors.DictCursor)

try:
    with connection.cursor() as cursor:
        # 创建KafkaData表
        cursor.execute("CREATE TABLE IF NOT EXISTS `KafkaData` (`id` int(11) NOT NULL AUTO_INCREMENT, `data` json NOT NULL, PRIMARY KEY (`id`))")

        # 写入数据到KafkaData表
        sql = "INSERT INTO `KafkaData` (`data`) VALUES (%s)"
        cursor.execute(sql, (json.dumps(data),))
        connection.commit()
finally:
    connection.close()

在这个示例中，我们首先使用pymysql.connect函数连接到MySQL数据库。然后，我们使用with语句创建一个数据库游标，并执行SQL语句来创建名为“KafkaData”的表。接下来，我们使用INSERT INTO语句将数据写入到表中，并使用json.dumps函数将JSON格式数据转换为字符串格式。

完整示例

以下是完成上述操作组合的完整示例代码：

from kafka import KafkaConsumer
import json
import pymysql.cursors

# 连接到Kafka服务器
consumer = KafkaConsumer('KafkaTest',
                         bootstrap_servers=['localhost:9092'],
                         auto_offset_reset='earliest',                                                  
                         enable_auto_commit=True,
                         group_id='json-group')

# 连接MySQL数据库
connection = pymysql.connect(host='localhost',
                             user='root',
                             password='password',
                             db='MyDB',
                             charset='utf8mb4',
                             cursorclass=pymysql.cursors.DictCursor)

try:
    with connection.cursor() as cursor:
        # 创建KafkaData表
        cursor.execute("CREATE TABLE IF NOT EXISTS `KafkaData` (`id` int(11) NOT NULL AUTO_INCREMENT, `data` json NOT NULL, PRIMARY KEY (`id`))")

        # 读取Kafka中的数据并写入到MySQL数据库
        for message in consumer:
            # 将数据解析为JSON格式
            data = json.loads(message.value)

            # 插入数据到KafkaData表
            sql = "INSERT INTO `KafkaData` (`data`) VALUES (%s)"
            cursor.execute(sql, (json.dumps(data),))

            # 提交并保存更改
            connection.commit()
finally:
    connection.close()

在这个示例中，我们首先连接到Kafka服务器，然后连接到MySQL数据库，在while循环中读取Kafka中的数据，将其解析为JSON格式，然后将其写入到MySQL中。