读取数据发送kafka

要将数据发送到Kafka，你需要使用Kafka的Producer API。下面是一个示例代码，假设你已经有了一个准备发送到Kafka的数据集：

from kafka import KafkaProducer
import json

producer = KafkaProducer(bootstrap_servers=['localhost:9092'], value_serializer=lambda m: json.dumps(m).encode('ascii'))

data = [
    {"name": "Alice", "age": 25},
    {"name": "Bob", "age": 30},
    {"name": "Charlie", "age": 35}
]

for d in data:
    producer.send('my_topic', d)

producer.flush()

在这个示例代码中，我们使用了kafka-python库中的KafkaProducer类来初始化一个生产者对象，并将其连接到本地运行的Kafka服务器。`value_serializer`参数指定了一个函数，用于将Python对象转换为JSON格式，以便于在Kafka中进行序列化和传输。

然后，我们将数据集中的每个数据都发送到了名为“my_topic”的Kafka主题中。

最后，我们使用`producer.flush()`函数来确保所有数据都已经发送到Kafka服务器，并且没有缓存留在生产者中。