Databricks中PySpark跨Notebook传递字典的性能问题咨询

核心结论 #

跨Notebook传递存储海量数据的字典完全不可行，必然引发严重性能问题甚至直接崩溃。原因很简单：字典是Driver节点的本地内存对象，3亿行数据全部拉到Driver内存会直接撑爆内存（OOM），完全违背Spark分布式计算的设计初衷。

可行处理方案 #

调整你的ETL流程，全程基于PySpark分布式DataFrame处理，替代字典传递的思路：

1. 优化数据源读取逻辑（保留DataFrame）

你的抽象类读取数据源的思路没问题，但要返回PySpark DataFrame而非字典——这是Spark处理大数据的正确姿势，数据会分布式存储在集群节点上，不会压垮Driver。

from abc import ABC, abstractmethod
import pyspark.sql.functions as F

class DataReader(ABC):
    @abstractmethod
    def read(self) -> pyspark.sql.DataFrame:
        pass

class CSVDataReader(DataReader):
    def __init__(self, path, header=True, inferSchema=True):
        self.path = path
        self.header = header
        self.inferSchema = inferSchema

    def read(self):
        # 根据实际需求配置读取参数（比如指定schema替代inferSchema提升性能）
        return spark.read.csv(self.path, header=self.header, inferSchema=self.inferSchema)

# 使用示例
csv_reader = CSVDataReader("/dbfs/path/to/300m_rows.csv")
df_source = csv_reader.read()

2. 跨Notebook共享逻辑/数据的正确方式

方式一：用%run导入转换逻辑（推荐）

把转换逻辑封装成独立Notebook，主Notebook通过%run直接导入函数，传递DataFrame和小配置字典（而非存数据的大字典）执行转换：

• 转换逻辑Notebook（data_transformations）：

def apply_transformations(df, source_config):
    # 通用转换逻辑：选列、过滤、数据清洗等
    transformed_df = df.select("id", "value", "create_time").filter(F.col("value").isNotNull())
    
    # 根据数据源配置执行差异化转换
    if source_config["type"] == "main_csv":
        transformed_df = transformed_df.withColumn("value_double", F.col("value") * 2)
    
    return transformed_df

• 主Notebook：

%run ./data_transformations

# 读取数据源
csv_reader = CSVDataReader("/dbfs/path/to/300m_rows.csv")
df_source = csv_reader.read()

# 传递小配置字典（仅存规则，不存数据）
source_config = {"type": "main_csv", "filter_threshold": 100}
transformed_df = apply_transformations(df_source, source_config)

方式二：用临时视图/DBFS存储共享中间数据

如果必须拆分多个独立执行的Notebook，可将中间DataFrame写入全局临时视图或DBFS Parquet文件，下一个Notebook直接读取：

• 读取数据Notebook：

# 写入全局临时视图（生命周期跟随SparkSession）
df_source.createOrReplaceGlobalTempView("temp_main_source")

# 或写入DBFS Parquet（持久化，适合长期存储）
df_source.write.mode("overwrite").parquet("/dbfs/temp/main_source_parquet")

• 转换Notebook：

# 从全局临时视图读取
df_source = spark.sql("SELECT * FROM global_temp.temp_main_source")

# 或从DBFS Parquet读取
df_source = spark.read.parquet("/dbfs/temp/main_source_parquet")

# 执行转换逻辑

3. 最终加载步骤

转换完成后直接将DataFrame写入Parquet并注册表：

# 写入Parquet（推荐分区存储提升后续查询性能）
transformed_df.write.mode("overwrite").partitionBy("create_time").parquet("/dbfs/output/target_table_parquet")

# 注册为Databricks表
spark.sql("""
CREATE TABLE IF NOT EXISTS target_table
USING PARQUET
LOCATION '/dbfs/output/target_table_parquet'
""")

关键注意事项 #

• 永远不要将分布式DataFrame转成本地集合（字典、列表等），这是Spark新手最容易踩的坑，会直接耗尽Driver内存。
• 处理3亿行数据时，务必指定明确的Schema（替代inferSchema），能大幅提升读取性能。
• 转换过程中尽量使用Spark内置函数（pyspark.sql.functions），避免用udf，减少性能损耗。

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