如何对DataFrame多列执行相同函数并并行化处理以更新参考表

需要基于输入数据更新参考表，实际业务列数更多，以下以3列示例说明需求逻辑：

• 输入表（表名input）结构：

• 参考表（表名references）存储各列对应单元格的取值、对应的md5哈希值以及所属列标识，结构：

现有处理函数逻辑：将输入表与参考表做比对，发现未收录的新数据时生成对应的参考记录，返回子DataFrame，最终用于更新参考表，现有3列场景的串行实现代码如下：

def repeatedly_excuted_funcion(input_data, references, col_name):
    """ 
    input_data: 全量输入DataFrame
    references: 待校验、更新的参考表
    col_name: 当前执行处理的目标列名
    """
    # ... 业务处理逻辑 ... 
    return partial_df

df_col1 = repeatedly_excuted_funcion(input_data, references, "col1")    
df_col2 = repeatedly_excuted_funcion(input_data, references, "col2")    
data_to_append = df_col1.union(df_col2)
df_col3 = repeatedly_excuted_funcion(input_data, references, "col3")
data_to_append = data_to_append.union(df_col2) 

注意：上述示例代码最后一行存在笔误，逻辑上应该合并df_col3而非重复合并df_col2

上述示例仅展示3列的处理逻辑，实际场景下共需要检查处理15列。最终目标是将新计算得到的md5值追加写入，完成参考表的更新，写入代码如下：

(
     data_to_append.write.format("delta")
     .mode("append")
     .saveAsTable(database_table)
)

批量处理多列（消除重复代码） #

不需要逐列手写调用逻辑，只需要维护目标列名列表，遍历执行即可，15列场景只需要把所有列名填入列表：

# 填入所有需要处理的15个列名
target_cols = ["col1", "col2", "col3", "col4", ..., "col15"]
# 批量执行所有列的处理
partial_dfs = [repeatedly_excuted_funcion(input_data, references, col) for col in target_cols]
# 合并所有列的处理结果
data_to_append = partial_dfs[0]
for df in partial_dfs[1:]:
    data_to_append = data_to_append.union(df)

并行化处理实现 #

从Delta写入逻辑判断当前使用的是PySpark环境，Spark本身的DataFrame是分布式懒执行，但逐列串行提交任务会导致执行计划串行排队。由于Spark任务实际调度运行在JVM侧，Python侧多线程不会受GIL锁限制，适合用来并行提交独立的列处理任务，实现方式如下：

from concurrent.futures import ThreadPoolExecutor, as_completed

# 填入所有需要处理的15个列名
target_cols = ["col1", "col2", "col3", "col4", ..., "col15"]
partial_dfs = []

# 线程数可根据集群资源调整，通常设为8~16，不要超过集群同时可运行的任务上限
with ThreadPoolExecutor(max_workers=8) as executor:
    # 提交所有列的处理任务
    future_to_col = {
        executor.submit(repeatedly_excuted_funcion, input_data, references, col): col
        for col in target_cols
    }
    # 收集执行完成的结果
    for future in as_completed(future_to_col):
        current_col = future_to_col[future]
        try:
            partial_dfs.append(future.result())
        except Exception as e:
            print(f"列{current_col}处理失败: {str(e)}")
            raise e

# 合并所有结果后写入
data_to_append = partial_dfs[0]
for df in partial_dfs[1:]:
    data_to_append = data_to_append.union(df)

注意事项 #

• 不要使用Python多进程实现并行，会触发SparkContext序列化报错，多线程是PySpark场景下提交独立任务的最优选择
• 如果处理函数内部包含shuffle类重计算逻辑，并行执行时注意监控集群资源，避免同时运行过多任务导致资源抢占拖慢整体速度
• 合并结果时可以用functools.reduce替代循环写法，效果等价：

  from functools import reduce
  data_to_append = reduce(lambda x, y: x.union(y), partial_dfs)
  

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