如何在DataBricks中并行读取文件？现有方案报错求优化

你的错误根源在于：SparkSession/SparkContext是绑定在Driver节点的对象，无法被序列化传递到Worker节点执行。当你用distPaths.map(processFile)时，processFile函数会被分发到Worker节点运行，但函数里的spark对象（依赖SparkContext）无法被序列化，所以抛出了PicklingError。

下面给你几个可行的优化方案，从简单到复杂：

方案1：让Spark原生并行读取（推荐） #

Spark的DataFrame API本身就支持并行读取多个文件，不需要手动用RDD的map来实现。如果你的需求是统计每个文件的记录数，可以结合input_file_name()函数来分组统计：

from pyspark.sql.functions import input_file_name

# 直接传入路径列表，Spark会自动并行读取所有文件
df = spark.read.json(paths)

# 添加文件名列，然后分组统计每个文件的记录数
file_count_df = df.withColumn("file_path", input_file_name()) \
                  .groupBy("file_path") \
                  .count()

# 收集结果并打印
for row in file_count_df.collect():
    print(f"文件 {row.file_path} 的记录数：{row['count']}")

这种方式的优势是利用Spark的优化引擎（Catalyst），性能比手动RDD操作更好，而且代码更简洁。

方案2：用mapPartitions处理复杂文件逻辑 #

如果需要对每个文件做更复杂的自定义处理（不止是count），可以用mapPartitions替代map。mapPartitions是对整个分区的路径批量处理，在每个分区的任务中，你可以通过SparkSession.getActiveSession()获取Worker节点上的Active SparkSession，避免序列化问题：

from pyspark.sql import SparkSession

def process_partition(paths_iter):
    # 获取当前Worker节点的Active SparkSession
    spark = SparkSession.getActiveSession()
    results = []
    for path in paths_iter:
        # 这里可以写自定义的文件处理逻辑，比如读取、转换等
        df = spark.read.json(path)
        results.append((path, df.count()))
    return results

# 并行化路径列表，然后用mapPartitions处理
dist_paths = sc.parallelize(paths)
file_counts = dist_paths.mapPartitions(process_partition).collect()

print("每个文件的记录数：")
for path, count in file_counts:
    print(f"{path}: {count}")

注意：mapPartitions是按分区处理，每个分区对应一个Worker任务，这样可以减少创建SparkSession的开销（每个分区只创建一次）。

额外优化建议 #

• 合并小文件：如果你的路径下有大量小文件，建议先合并成大文件，避免Spark启动过多任务，降低性能。可以用df.repartition(n).write.mode("overwrite").json(output_path)来合并。
• 自动遍历目录：如果文件都在同一个目录下，可以用spark.read.json("/path/to/directory")，或者添加option("recursiveFileLookup", "true")来递归读取子目录的文件，不用手动列路径。
• 优先使用DataFrame API：DataFrame API比RDD有更多的优化，尽量避免手动RDD操作，除非有特殊的自定义需求。

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