Python递归生成迭代JSON及Avro写入重复数据问题求助

你的核心问题出在get_test函数的实现上：

• 每次调用get_test都会重新打开test.csv，从文件开头读取内容
• 函数内部的循环在第一次迭代时就执行return，直接返回第一行数据，后续行根本不会被处理
• 因此avro_gen里循环调用get_test时，每次拿到的都是文件第一行的内容，导致写入Avro的内容全部重复

方案一：将get_test改为生成器函数 #

生成器可以逐行返回CSV数据，既避免重复打开文件，也能遍历所有行：

import csv
from avro.datafile import DataFileWriter
from avro.io import DatumWriter
import avro.schema

def get_test():
    with open("test.csv", 'r') as f:
        reader = csv.DictReader(f)
        # 遍历所有行，逐行生成转换后的字典
        for row in reader:
            yield {
                "field1": row['field1'],
                "field2": bool(row['field2']),
            }

def avro_gen():
    schema = avro.schema.parse(open("test.avsc", "rb").read())
    with open("test_.avro", 'wb') as f:
        writer = DataFileWriter(f, DatumWriter(), schema)
        # 直接遍历生成器，自动处理所有行，无需提前统计行数
        for data in get_test():
            writer.append(data)
        writer.close()

方案二：将CSV读取逻辑整合到avro_gen中 #

如果不需要单独复用CSV读取逻辑，直接在生成Avro的函数里处理CSV会更高效：

import csv
from avro.datafile import DataFileWriter
from avro.io import DatumWriter
import avro.schema

def avro_gen():
    schema = avro.schema.parse(open("test.avsc", "rb").read())
    # 同时打开CSV和Avro文件，减少IO操作
    with open("test.csv", 'r') as csv_file, open("test_.avro", 'wb') as avro_file:
        reader = csv.DictReader(csv_file)
        writer = DataFileWriter(avro_file, DatumWriter(), schema)
        for row in reader:
            data = {
                "field1": row['field1'],
                "field2": bool(row['field2']),
            }
            writer.append(data)
        writer.close()

关键说明 #

• 两种方案都避免了重复打开CSV文件的操作，一次性读取并处理所有行
• 方案一的生成器可以方便复用CSV转字典的逻辑，适合需要多处使用该转换的场景
• 方案二更简洁高效，适合只在生成Avro时使用的场景

内容的提问来源于stack exchange，提问作者Nithin Thomas