使用PyArrow批量添加列报错：RecordBatch无append_column属性

错误原因 #

你遇到的AttributeError是因为PyArrow的RecordBatch是不可变结构，它没有append_column方法——这个方法仅存在于pyarrow.Table类中。此外你原批量代码还有一个问题：每次调用pq.write_table会直接覆盖输出文件，最终只会保留最后一个Batch的内容。

可行解决方法 #

方法一：收集所有处理后的Batch，合并为Table后一次性写入 #

先逐个处理每个RecordBatch，将处理后的Batch收集起来，最后合并成完整Table再写入磁盘，避免文件被多次覆盖：

import pyarrow.parquet as pq
import pyarrow as pa

def function(row):
    # 这里是你的行处理逻辑，示例返回一个计算值
    return row.get('some_column', 0) * 2

# 读取原始Table
candidates = pq.ParquetFile('input.parquet').read()
processed_batches = []

for batch in candidates.to_batches():
    result = []
    for row in batch.to_pylist():
        row_result = function(row)
        result.append(row_result)
    # 将RecordBatch转为Table，添加列后再转回RecordBatch
    batch_table = batch.to_table()
    batch_table_with_col = batch_table.append_column('new_column_name', pa.array(result))
    processed_batches.append(batch_table_with_col.to_batches()[0])

# 合并所有处理后的Batch为一个Table，写入磁盘
final_table = pa.Table.from_batches(processed_batches)
pq.write_table(final_table, 'output.parquet')

方法二：使用ParquetWriter逐个追加处理后的Batch #

如果原始文件过大，不想一次性加载到内存，可以用pq.ParquetWriter逐个追加处理后的Batch到输出文件：

import pyarrow.parquet as pq
import pyarrow as pa

def function(row):
    # 你的行处理逻辑
    return row.get('some_column', 0) * 2

# 读取原始Table并扩展Schema，添加新列的字段定义
candidates = pq.ParquetFile('input.parquet').read()
new_schema = candidates.schema.append(pa.field('new_column_name', pa.int64())) # 根据实际数据类型调整

# 初始化ParquetWriter，逐个写入处理后的Batch
with pq.ParquetWriter('output.parquet', new_schema) as writer:
    for batch in candidates.to_batches():
        result = []
        for row in batch.to_pylist():
            row_result = function(row)
            result.append(row_result)
        # 构造包含新列的RecordBatch
        new_columns = batch.columns + [pa.array(result)]
        new_batch = pa.RecordBatch.from_arrays(new_columns, schema=new_schema)
        writer.write_batch(new_batch)

额外优化建议 #

如果你的function可以通过矢量化方式实现（比如用PyArrow的compute函数），建议避免转成pylist循环，能大幅提升处理性能。示例：

import pyarrow.compute as pc
# 用矢量化计算替代行循环
new_column = pc.multiply(candidates['some_column'], 2)
candidates_with_result = candidates.append_column('new_column_name', new_column)

内容的提问来源于stack exchange，提问作者Dino Burger