Python从Oracle导出Avro消息：无需反序列化直接写入文件可行吗？

我在应用中使用Oracle数据库存储代表Kafka事件的不同Schema的Avro消息，消息主体存在单表的CLOB列BODY中，现在需要把这些消息导出为合法的Avro文件。

使用以下Python代码从数据库取数时：

# Return string or bytes instead of a locator (for CLOBs e.g. event.body)
oracledb.defaults.fetch_lobs = False

...

cursor.execute(
    'select body from EVENTS' +
    (f" where EVENT_TYPE='{event_type}'" if event_type else "") +
    " order by CREATED_ON"
)
for body in cursor:
    filename = fr'./exported_events/event_{SOME_UNIQUE_ID}.avro'
    os.makedirs(os.path.dirname(filename), exist_ok=True)
    with open(filename, 'wb+') as file:
        fastavro.writer(
                file,
                fastavro_schema.load_schema(fr"./event_{event_type}.avsc"),
                [body]
        )

出现错误：提示body是bytes对象无法迭代。看起来需要先序列化再反序列化才能写入Avro文件，有没有办法避免这个操作？

另外，我试过用json.dumps(body)把body转成字典，但消息里的bytearrays和timestamp-millis类型会被错误转换，导致Schema验证报错。

解决方案 #

1. 先解决bytes对象无法迭代的核心问题 #

你拿到的body其实是元组类型（cursor迭代返回的每行数据都是元组，哪怕只查询了一列），也就是(bytes_data,)，而不是单纯的bytes。把元组放进列表[body]后，fastavro会尝试迭代元组内的元素，进而把bytes按单个字节拆分，这才引发了错误。

修正取数逻辑即可：

for row in cursor:
    body = row[0]  # 提取元组中第一个元素，即CLOB对应的bytes数据
    # 后续写入逻辑

2. 直接导出原始Avro二进制（无需序列化/反序列化） #

如果你的CLOB中存储的是Kafka传输的原始Avro二进制消息（包含完整的Avro序列化数据），那完全不需要用fastavro.writer处理，直接将bytes写入文件就行：

for row in cursor:
    body = row[0]
    filename = fr'./exported_events/event_{SOME_UNIQUE_ID}.avro'
    os.makedirs(os.path.dirname(filename), exist_ok=True)
    with open(filename, 'wb') as file:
        file.write(body)

注意：这种方式导出的是单条Avro消息的二进制文件；如果需要生成包含多条记录的Avro容器文件，还是得走反序列化再写入的流程。

3. 正确处理类型的序列化/反序列化流程 #

如果必须生成标准Avro容器文件，那就得用对应的Schema把CLOB中的bytes反序列化为Python对象，再用fastavro写入。这种方式能完整保留bytearrays和timestamp-millis类型：

import io
import fastavro

# 提前加载对应Schema
schema = fastavro_schema.load_schema(fr"./event_{event_type}.avsc")

for row in cursor:
    body_bytes = row[0]
    # 用Schema解析原始Avro二进制
    record = fastavro.schemaless_reader(io.BytesIO(body_bytes), schema)
    
    filename = fr'./exported_events/event_{SOME_UNIQUE_ID}.avro'
    os.makedirs(os.path.dirname(filename), exist_ok=True)
    with open(filename, 'wb+') as file:
        fastavro.writer(file, schema, [record])

4. 额外修复：避免SQL注入风险 #

你的代码用字符串拼接生成SQL，存在SQL注入漏洞，建议改用绑定变量：

if event_type:
    cursor.execute(
        'select body from EVENTS where EVENT_TYPE = :event_type order by CREATED_ON',
        {'event_type': event_type}
    )
else:
    cursor.execute('select body from EVENTS order by CREATED_ON')

内容的提问来源于stack exchange，提问作者Hercules Konsoulas