如何用Python从kafka读取数据并写入Oracle？

你可以使用 Kafka-Python 和 cx_Oracle 这两个库来实现从 Kafka 读取数据并将数据写入 Oracle 数据库。

以下是一个简单的 Python 示例代码：

from kafka import KafkaConsumer
import cx_Oracle

# Kafka 配置
kafka_topic = 'my_topic'
kafka_bootstrap_servers = ['localhost:9092']

# Oracle 配置
oracle_user = 'my_user'
oracle_password = 'my_password'
oracle_host = 'localhost'
oracle_port = '1521'
oracle_service_name = 'my_service_name'

# Kafka 消费者
consumer = KafkaConsumer(kafka_topic,
                         bootstrap_servers=kafka_bootstrap_servers,
                         auto_offset_reset='earliest',
                         enable_auto_commit=True,
                         group_id='my_group')

# Oracle 连接
dsn = cx_Oracle.makedsn(oracle_host, oracle_port, service_name=oracle_service_name)
conn = cx_Oracle.connect(user=oracle_user, password=oracle_password, dsn=dsn)
cursor = conn.cursor()

# 读取 Kafka 消息并写入 Oracle
for message in consumer:
    # 解析消息
    data = message.value.decode('utf-8').split(',')
    id = int(data[0])
    name = data[1]
    age = int(data[2])
    # 写入 Oracle
    cursor.execute("INSERT INTO my_table VALUES (:id, :name, :age)", [id, name, age])
    conn.commit()

# 关闭连接
cursor.close()
conn.close()

在这个示例中，我们首先配置了 Kafka 和 Oracle 的连接参数，并创建了一个 Kafka 消费者和一个 Oracle 连接。然后我们使用一个 `for` 循环从 Kafka 主题中读取消息，并将消息解析为数据。最后，我们使用 `cx_Oracle` 库将数据插入到 Oracle 数据库中。注意，我们在处理完每条消息后都需要手动提交事务，以确保数据被正确存储到数据库中。

当然，这只是一个简单的示例。在实际情况下，你可能需要考虑更多的细节，例如异常处理、数据验证、批量写入等等。