Kafka流追加至JSON作为事件丰富化

要将Kafka流追加至JSON以丰富事件，你可以使用以下代码示例：

首先，你需要安装Kafka Python库，你可以使用以下命令安装它：

pip install kafka-python

接下来，你可以使用以下代码将Kafka流追加至JSON：

from kafka import KafkaConsumer
import json

# 创建Kafka消费者
consumer = KafkaConsumer('topic_name', bootstrap_servers=['localhost:9092'])

# 将Kafka流追加至JSON文件
with open('output.json', 'a') as json_file:
    for message in consumer:
        # 解析Kafka消息的键值对
        key = message.key.decode('utf-8')
        value = json.loads(message.value.decode('utf-8'))
        
        # 将消息转换为JSON格式并写入文件
        json_data = {
            'key': key,
            'value': value
        }
        json_file.write(json.dumps(json_data) + '\n')

在上述代码中，我们使用KafkaConsumer从指定的主题（topic）创建了一个Kafka消费者。然后，我们使用一个循环来迭代消费者收到的每个消息。接下来，我们解析了消息的键值对，并将其转换为JSON格式。最后，我们将JSON数据写入名为output.json的文件中。

请注意，上述代码中的'topic_name'应替换为你要消费的实际主题名称。另外，你也可以根据自己的需求修改输出JSON的文件名和路径。

注意：在实际使用中，你可能需要添加更多的错误处理和异常处理来处理连接错误、解析错误等情况。此示例仅提供了基本的代码结构。