Spark/Databricks并行加载多城市CSV到对应表的优化方案问询

你当前的串行代码是在Driver端单线程逐个执行城市文件的读、写操作，所有任务排队执行，完全浪费了分布式集群的并行资源，下面提供两种生产可用的并行改造方案：

方案1：Spark RDD分布式并行（推荐，性能最优） #

利用Spark原生的分布式并行能力，把城市列表转成RDD分发到Executor节点并行处理，无需额外依赖，性能最高：

staging_path = "abfss://xyz/abc/"

# Driver端提前提取所有城市名，避免Executor端调用dbutils出现权限问题
filesProp = dbutils.fs.ls(staging_path)
city_set = set()
for file in filesProp:
    # 修正原代码的分隔符bug：文件名用下划线分隔，不是短横线
    city_name = file.name.split('_')[0]
    city_set.add(city_name)
city_list = list(city_set)

# 单城市处理逻辑，会分发到Executor端独立执行
def process_single_city(city_name):
    file_path = f"{staging_path}{city_name}_*.csv"
    # 读取对应城市的所有CSV文件，可提前传入对应城市的Schema避免推断错误
    city_df = spark.read.options(header='True', delimiter=',').csv(file_path)
    # 写入表，需要覆盖旧数据可替换为mode("overwrite")
    city_df.write.mode("append").saveAsTable(city_name)
    return f"{city_name} 表创建完成"

# 并行度设置为城市总数100，可根据集群资源调整
spark.sparkContext.parallelize(city_list, numSlices=100).foreach(process_single_city)

方案2：Python多线程并行（改造成本最低） #

如果不想调整原有代码结构，可以用Python线程池在Driver端并行提交Spark任务，适合中小集群快速改造：

from concurrent.futures import ThreadPoolExecutor, as_completed

staging_path = "abfss://xyz/abc/"

# 提前提取城市列表
filesProp = dbutils.fs.ls(staging_path)
city_set = set()
for file in filesProp:
    city_name = file.name.split('_')[0]
    city_set.add(city_name)
city_list = list(city_set)

# 单城市处理逻辑和原逻辑基本一致
def process_single_city(city_name):
    file_path = f"{staging_path}{city_name}_*.csv"
    city_df = spark.read.options(header='True', delimiter=',').csv(file_path)
    city_df.write.mode("append").saveAsTable(city_name)
    return f"{city_name} 处理完成"

# 最大并行数根据集群CPU、内存资源调整，建议设置为10~20避免任务过载
max_workers = 15
with ThreadPoolExecutor(max_workers=max_workers) as executor:
    futures = [executor.submit(process_single_city, city) for city in city_list]
    for future in as_completed(futures):
        print(future.result())

注意事项 #

• 修正原代码的两个bug：一是文件名分隔符用错，二是dbutils.fs.ls传入的变量名和定义的staging_path不一致
• 如果需要严格控制每个城市的表结构，建议提前把所有城市的Schema存在字典中，读CSV时显式传入schema参数，避免Spark自动推断Schema出错
• 数据量极大的场景优先选方案1，改造成本优先选方案2

内容的提问来源于stack exchange，提问作者Surendranatha Reddy Chappidi