Spark DataFrame遍历及循环中列表修改问题求助

问题描述 #

我正在开发一个Python函数，流程如下：

• 遍历存储数据结构信息的Pandas DataFrame，获取目标字段对应的表信息
• 加载对应Spark表并遍历其DataFrame
• 匹配到目标字段值时，将其添加到记录列表与最终DataFrame，最后导出为CSV

当前代码使用collect()处理Spark DataFrame，但调用函数后传入的空列表liste_t始终无返回值。想解决两个问题：

1. 如何正确遍历Spark DataFrame
2. 如何在循环中修改并返回列表/DataFrame（担心循环内的修改无法作用到外部）

附相关代码：

df_meta = pd.read_csv("/dbfs/mnt/resources/path/file_meta.csv", sep=';')
liste_t = []

def recursive_process(field, id_p, list_drop): 
    for row in df_meta.index:
        if df_meta['SOURCE_COLUMN_NAME'][row] == field:
            df_table = spark.read.table("source1"+"."+df_meta['SOURCE_TABLE_NAME'][row])
            data_collect = df_table.collect()
        for row2 in data_collect:
            if row2(field) == id_p and row2(field) not in list_drop:
                list_drop.append(id_p)
                #add field + value to final dataframe
    return list_drop

调用方式：

recursive_process("Col_ID","1003729193",liste_t)

问题分析与解决方案 #

1. 原代码的核心问题 #

• 变量未初始化风险：data_collect仅在df_meta匹配字段时赋值，若循环遇到不匹配的行，data_collect未定义会直接报错。
• Spark行访问错误：Spark DataFrame的行对象不能用row2(field)调用，正确写法是row2[field]（按列名索引）或row2.field（属性访问）。
• 效率与逻辑漏洞：直接collect()拉取全表数据，不仅效率低还易引发内存溢出；判断条件中row2(field) == id_p和row2(field) not in list_drop属于冗余判断，只需校验id_p是否在列表即可。
• 递归未实现：函数名标注为recursive_process但无递归调用逻辑，不符合流程预期。

2. 正确遍历Spark DataFrame的方式 #

Spark是分布式计算框架，优先用Spark API做过滤、转换，避免直接collect()拉取全表：

• 用filter()提前过滤出目标字段等于id_p的行，减少数据量
• 如需遍历，用toLocalIterator()替代collect()，逐批次拉取数据，节省内存

3. 列表修改的正确性 #

Python中列表是可变对象，函数内对列表的append等修改操作会直接作用于外部传入的列表，无需额外处理，只要逻辑正确就能触发修改。

4. 修正后的代码示例 #

import pandas as pd

df_meta = pd.read_csv("/dbfs/mnt/resources/path/file_meta.csv", sep=';')
liste_t = []

def recursive_process(field, id_p, list_drop): 
    # 遍历元数据表，用iterrows()更直观
    for idx, row_meta in df_meta.iterrows():
        if row_meta['SOURCE_COLUMN_NAME'] == field:
            table_name = f"source1.{row_meta['SOURCE_TABLE_NAME']}"
            df_table = spark.read.table(table_name)
            
            # Spark API过滤目标数据，避免全表拉取
            filtered_df = df_table.filter(df_table[field] == id_p)
            
            # 用toLocalIterator()遍历，节省内存
            for row in filtered_df.toLocalIterator():
                current_id = row[field]
                if current_id not in list_drop:
                    list_drop.append(current_id)
                    # 此处可添加字段值到最终存储结构，比如维护一个记录列表
    
    # 补充递归逻辑（根据实际需求调整，比如处理关联字段）
    # for related_field in get_related_fields(field):
    #     recursive_process(related_field, id_p, list_drop)
    
    return list_drop

调用方式不变，调用后liste_t会直接被修改：

recursive_process("Col_ID","1003729193",liste_t)
print(liste_t)

5. 最终导出CSV的建议 #

维护一个记录列表存储所有匹配数据，最后转换为Spark/Pandas DataFrame导出：

# 修改函数，同时收集结果记录
def recursive_process(field, id_p, list_drop, result_records): 
    for idx, row_meta in df_meta.iterrows():
        if row_meta['SOURCE_COLUMN_NAME'] == field:
            table_name = f"source1.{row_meta['SOURCE_TABLE_NAME']}"
            df_table = spark.read.table(table_name)
            filtered_df = df_table.filter(df_table[field] == id_p)
            
            for row in filtered_df.toLocalIterator():
                current_id = row[field]
                if current_id not in list_drop:
                    list_drop.append(current_id)
                    # 收集整行数据，转换为字典
                    result_records.append(row.asDict())
    
    # 递归逻辑（按需添加）
    return list_drop, result_records

# 使用示例
result_records = []
recursive_process("Col_ID","1003729193",liste_t, result_records)

# Spark DataFrame导出（适合大数据量）
spark.createDataFrame(result_records).write.csv("/dbfs/mnt/output/result.csv", header=True, sep=';')
# Pandas DataFrame导出（适合小数据量）
pd.DataFrame(result_records).to_csv("/dbfs/mnt/output/result_pandas.csv", header=True, sep=';', index=False)

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