Avro DataFileReader适配Python3标准输入失败，求解决方案

我之前在Python 3里处理Avro和stdin的组合时，也踩过几乎一模一样的坑，给你几个针对性的解决思路：

1. 用二进制模式读取stdin并包装为可seek的BytesIO #

Python 3里的sys.stdin默认是文本模式，会自动尝试用UTF-8解码输入内容，但Avro数据文件是二进制格式，开头的Obj\x01标识根本不是合法的UTF-8字节，这就是你碰到UnicodeDecodeError的核心原因。

解决的关键是先以二进制模式读取stdin，再用io.BytesIO把读取到的内容包装成支持seek操作的对象：

import sys
import io
from avro.datafile import DataFileReader
from avro.io import DatumReader

def process_avro_from_stdin():
    # 读取stdin的原始二进制数据
    raw_data = sys.stdin.buffer.read()
    # 包装成可seek的BytesIO对象
    seekable_buffer = io.BytesIO(raw_data)
    
    try:
        reader = DataFileReader(seekable_buffer, DatumReader())
        for record in reader:
            # 这里替换成你的业务逻辑
            print(record)
        reader.close()
    except Exception as e:
        print(f"处理失败: {str(e)}")

if __name__ == "__main__":
    process_avro_from_stdin()

这个方案适合数据量不大的场景，所有数据会加载到内存中处理。

2. 用临时文件处理大体积输入 #

如果stdin传输的Avro数据很大，全部加载到内存会导致资源占用过高，这时候可以用临时文件来存储数据——临时文件天然支持seek操作：

import sys
import tempfile
from avro.datafile import DataFileReader
from avro.io import DatumReader

def process_large_avro_from_stdin():
    # 创建一个可读写的临时文件（二进制模式），程序结束后自动删除
    with tempfile.NamedTemporaryFile(mode='wb+', delete=True) as temp_file:
        # 将stdin的二进制数据写入临时文件
        temp_file.write(sys.stdin.buffer.read())
        # 回到文件开头，准备读取
        temp_file.seek(0)
        
        try:
            reader = DataFileReader(temp_file, DatumReader())
            for record in reader:
                # 处理你的记录
                print(record)
            reader.close()
        except Exception as e:
            print(f"处理失败: {str(e)}")

if __name__ == "__main__":
    process_large_avro_from_stdin()

3. 验证输入数据的完整性 #

如果上面的方案还是报avro.schema.AvroException: Not an Avro data file，那大概率是stdin传入的内容不是完整的Avro DataFile。你可以先把stdin内容保存到本地文件，用Avro官方工具验证：

# 将stdin内容保存到文件
cat /dev/stdin > test.avro
# 用avro-tools验证并转换为JSON（需提前安装avro-tools）
avro-tools tojson test.avro

如果工具也报错，说明输入的数据本身有问题（比如传输截断、文件损坏），这时候需要从数据源头排查问题。

内容的提问来源于stack exchange，提问作者Misha AM