Great Expectations在PySpark大数据集上性能低下的排查问询

我们在基于PySpark构建的ETL流水线中集成Great Expectations做数据质量校验，小数据集表现正常，但处理350GB Delta格式数据集时，仅对6列做非空值校验就耗时1小时30分钟。以下是我们尝试的两种实现方式，二者耗时相近，想请教是否存在操作不当之处？

第一种实现方式 #

from great_expectations.dataset.sparkdf_dataset import SparkDFDataset

dq_cols = ["col_1", "col_2", "col_3", "col_4", "col_5", "col_6"]
df = spark.read.load(...)
ge_df = SparkDFDataset(df)
for col in dq_cols:
    ge_df.expect_column_values_to_not_be_null(col)
ge_df.validate()

第二种实现方式 #

df = spark.read.load(...)
context = gx.get_context()
dataframe_datasource = context.sources.add_or_update_spark(name="my_spark_name")
dataframe_asset = dataframe_datasource.add_dataframe_asset(name="my_dataframe_asset_name",dataframe=df)
batch_request = dataframe_asset.build_batch_request()
expectations_object = load_yaml_expectations("...") # 自定义函数，读取包含校验规则的yaml文件
suite = context.add_or_update_expectation_suite(
        expectation_suite_name="my_expectation_suite_name",
        expectations=expectations_object,
    )
checkpoint = Checkpoint(
        name="my_checkpoint_name",
        # run_name_template="my_template_name",
        data_context=context,
        batch_request=batch_request,
        expectation_suite_name="my_expectation_suite_name",
        action_list=[
            {
                "name": "validate_expectations",
                "action": {"class_name": "StoreValidationResultAction"},
            }
        ],
    )

result = checkpoint.run()

1. 合并全表扫描操作 #

当前两种方式都会为每列非空校验触发一次独立的全表扫描，6列就会扫6次350GB数据，这是核心性能瓶颈。

• 改用一次性聚合校验：先通过Spark计算所有目标列的空值统计，再基于聚合结果在GE中完成校验，只需要一次全表扫描。
• 尝试使用表级别校验expect_table_columns_to_not_be_null，部分GE版本会自动优化为单次扫描逻辑。

2. 利用Delta Lake特性减少扫描量 #

• 生成并更新Delta表的统计信息，让Spark能智能跳过不必要的数据：

ANALYZE TABLE delta.`path/to/your/table` COMPUTE STATISTICS FOR ALL COLUMNS;

• 如果是分区表，读取时必须指定分区过滤条件，GE会继承Spark的分区下推优化，大幅减少扫描数据量。

3. 调整GE数据源配置 #

避免直接传入已加载的DataFrame给GE，改用Delta Lake直接数据源，减少额外的DataFrame转换开销：

context = gx.get_context()
delta_datasource = context.sources.add_or_update_delta(name="delta_source", spark=spark)
delta_asset = delta_datasource.add_delta_table_asset(name="delta_asset", table_name="your_table_name")
batch_request = delta_asset.build_batch_request()

4. 精简校验结果输出 #

如果不需要详细的校验明细（比如空值的具体行），在expectation中设置精简的结果格式，减少GE的IO和计算开销：

ge_df.expect_column_values_to_not_be_null(col, result_format="BASIC")

若不需要持久化校验结果，可临时关闭StoreValidationResultAction。

5. 优化Spark集群资源 #

• 根据数据量调整集群节点配置：350GB数据建议使用至少8核16G以上的节点，节点数量按需扩容。
• 调整Spark参数：设置spark.sql.shuffle.partitions为核心数的2-4倍，开启spark.sql.adaptive.enabled让Spark自动优化分区。

6. 升级GE版本 #

确认使用的GE版本对Spark 3.x和Delta Lake有优化，旧版本可能存在大数据集性能bug，建议升级到0.17.x及以上的稳定版。

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