帮我写一份消费kafka写入timelyre的样例代码

在Python中，你可以使用`confluent-kafka`库来操作Kafka，并结合`timely`库处理时间序列数据。这里是一个简单的示例，展示如何从Kafka主题读取消息并将其写入到`timely`的时间序列数据库：

from confluent_kafka import Producer
import timely
import json

# 设置Kafka生产者配置
producer_config = {
    'bootstrap.servers': 'localhost:9092',  # 替换为你本地或集群的Kafka服务器地址
    'key.serializer': 'utf_8',
    'value.serializer': lambda v: json.dumps(v).encode('utf-8')
}

# 初始化Kafka producer
producer = Producer(producer_config)

# 设定Timely的表名
table_name = "your_table_name"

# 创建Timely连接
with timely.Timely() as db:
    for msg in consumer.poll(1):  # 消费Kafka主题的消息
        if msg.error():
            print(f"Failed to consume message: {msg.error()}")
            continue
        
        data = json.loads(msg.value().decode('utf-8'))  # 解析消息内容为字典
        event_time = timely.now()  # 获取当前事件时间
        
        # 将数据写入Timely表
        db.insert(table_name, {"data": data}, timestamp=event_time)

        # 发送消息到Kafka
        producer.produce(topic='your_kafka_topic', key=str(event_time), value=json.dumps(data).encode('utf-8'))

        # 提交批量发送
        producer.flush()

# 关闭producer
producer.close()

在这个例子中，你需要替换`localhost:9092`为你的Kafka服务的实际地址，以及`your_table_name`和`your_kafka_topic`为你要使用的实际表名和Kafka主题。记得安装必要的库 (`confluent-kafka` 和 `timely`)。