Polars处理CSV内存占用高、耗时久，如何优化？

我正在执行以下操作：

• 使用Polars惰性数据帧（lazy dataframe）扫描CSV文件
• 通过函数格式化电话号码为E164格式
• 移除空值与重复数据
• 将结果写入新CSV文件

我的代码如下：

import sys
import json
import polars as pl
import phonenumbers

# 解析传入的JSON参数
args = json.loads(sys.argv[1])

# 格式化电话号码为E164格式
def parse_phone_number(phone_number):
    try:
        return phonenumbers.format_number(phonenumbers.parse(phone_number, "US"), phonenumbers.PhoneNumberFormat.E164)
    except phonenumbers.NumberParseException:
        pass
    return None

# 扫描CSV、处理数据并写入新文件
pl.scan_csv(args['path'], separator=args['delimiter']).select(
    [args['column']]
).with_columns(
    # 将整数手机号转为字符串并应用格式化函数
    pl.col(args['column']).cast(pl.String).map_elements(parse_phone_number).alias(args['column']),
    # 添加list_id列，值为传入的参数
    pl.lit(args['list_id']).alias("list_id")
    
).filter(
    # 过滤空值
    pl.col(args['column']).is_not_null()
).unique(keep="last").collect().write_csv(args['saved_path'], separator=",")

我测试了一个含80万行、23列（文件大小150MB）的数据集，完成任务耗时约20秒，内存占用超500MB。请问这种情况是否正常？能否优化性能（重点降低内存占用）？

注：我刚接触Polars，日常使用PHP，Python也很生疏，若代码有不足请见谅。

1. 现状是否正常？ #

这个表现正常但存在明显优化空间：

• 内存占用高：collect()会把处理后的全量数据集加载到内存，加上map_elements逐行调用Python函数产生的中间状态、unique操作需要全量数据比对，导致内存峰值飙升。
• 耗时较长：核心瓶颈是map_elements的逐行Python函数调用——Python循环的开销远大于Polars原生的矢量化操作。

2. 优化方案（降内存+提速度） #

（1）用map_batches替代map_elements，减少Python交互开销 #

map_elements是逐行处理，map_batches按批次处理数据，能大幅降低Python与Polars的交互次数。修改格式化函数为批量处理：

def parse_phone_numbers_batch(phone_numbers):
    results = []
    for num in phone_numbers:
        try:
            results.append(phonenumbers.format_number(phonenumbers.parse(num, "US"), phonenumbers.PhoneNumberFormat.E164))
        except phonenumbers.NumberParseException:
            results.append(None)
    return results

然后替换with_columns中的对应代码：

pl.col(args['column']).cast(pl.String).map_batches(parse_phone_numbers_batch).alias(args['column']),

（2）读取CSV时直接指定目标列，避免加载冗余数据 #

原文件有23列，但你只需要其中1列，在scan_csv阶段就指定只读取目标列，彻底跳过其他列的加载：

pl.scan_csv(
    args['path'], 
    separator=args['delimiter'],
    columns=[args['column']]  # 仅加载需要的列
)

这一步能直接砍掉22列数据的内存占用，效果非常明显。

（3）用sink_csv替代collect().write_csv，避免全量内存缓存 #

sink_csv支持边处理边写入文件，不需要把整个数据集加载到内存，能大幅降低内存峰值：

pl.scan_csv(
    args['path'], 
    separator=args['delimiter'],
    columns=[args['column']]
).with_columns(
    pl.col(args['column']).cast(pl.String).map_batches(parse_phone_numbers_batch).alias(args['column']),
    pl.lit(args['list_id']).alias("list_id")
).filter(
    pl.col(args['column']).is_not_null()
).unique(keep="last").sink_csv(args['saved_path'], separator=",")

（4）优化CSV读取参数 #

根据数据特性添加以下参数，进一步降低内存占用：

pl.scan_csv(
    args['path'], 
    separator=args['delimiter'],
    columns=[args['column']],
    dtypes={args['column']: pl.String},  # 若原手机号是整数，直接按字符串读取，跳过后续cast操作
    low_memory=True  # 按列分块读取，降低内存峰值
)

优化后预期效果 #

• 内存占用：可降至100MB以内（仅加载1列+边处理边写入）
• 耗时：可压缩至10秒以内（批量处理替代逐行处理）

内容的提问来源于stack exchange，提问作者Sarwar Alam