Great Expectations 与 Spark Connect 兼容性问题分析与解决方案

背景介绍

Great Expectations 是一个开源的数据质量验证工具，它能够帮助数据团队定义、记录和验证数据质量预期。在最新版本中，用户在使用 Spark Connect 数据框架时遇到了兼容性问题，导致无法正常构建批处理请求。

问题现象

当用户尝试使用 Spark 3.5.0 及以上版本中的 Spark Connect DataFrame 时，Great Expectations 1.0.0 版本会抛出错误："Bad input to build_batch_request: Can not build batch request for dataframe asset without a dataframe"。这是因为新版本 Spark 改变了 DataFrame 的导入路径，从原来的 pyspark.sql.DataFrame 变为了 pyspark.sql.connect.dataframe.DataFrame。

技术分析

根本原因

Great Expectations 在 spark_datasource.py 文件中通过 isinstance() 检查 DataFrame 类型时，仅检查了传统的 pyspark.sql.DataFrame 类型，而没有考虑到 Spark Connect 引入的新 DataFrame 类型。这种严格的类型检查导致了兼容性问题。

版本差异

在 Great Expectations 0.18.19 版本中，这个问题不存在，因为当时的实现方式更加宽松，没有进行严格的类型检查。但随着 1.0.0 版本的重构，引入了更严格的类型验证机制，暴露了与 Spark Connect 的兼容性问题。

解决方案

官方修复

Great Expectations 团队在 1.0.5 版本中修复了这个问题，修改了类型检查逻辑，使其同时支持传统的 pyspark.sql.DataFrame 和新的 pyspark.sql.connect.dataframe.DataFrame 类型。用户只需升级到最新版本即可解决此问题。

迁移指南

对于从旧版本迁移的用户，需要注意以下几点变化：

1. 上下文初始化：不再使用 BaseDataContext，而是使用 gx.get_context()
2. 期望配置：ExpectationConfiguration 的使用方式有所变化
3. 验证流程：新的验证流程更加简洁，直接通过 batch.validate() 方法执行验证

最佳实践

新版本代码示例

以下是使用 Great Expectations 1.0.5+ 验证 Spark DataFrame 的推荐方式：

import great_expectations as gx
from pyspark.sql import SparkSession

# 初始化Spark会话
spark = SparkSession.builder.appName("GEExample").getOrCreate()

# 创建示例DataFrame
data = [("2023-01-10", "A", 40), ("2023-05-15", "A", 60)]
df = spark.createDataFrame(data, ["date", "product", "value"])

# 初始化Great Expectations上下文
context = gx.get_context()

# 配置数据源和数据资产
data_source = context.data_sources.add_spark(name="my_source")
data_asset = data_source.add_dataframe_asset(name="my_asset")

# 创建批处理定义
batch_def = data_asset.add_batch_definition_whole_dataframe("my_batch")
batch_params = {"dataframe": df}

# 获取批处理对象
batch = batch_def.get_batch(batch_parameters=batch_params)

# 定义并执行验证
expectation = gx.expectations.ExpectColumnValuesToBeBetween(
    column="value", min_value=50, max_value=100
)
results = batch.validate(expectation)

处理JSON期望配置

对于需要从JSON加载期望配置的场景，可以使用以下方式：

from great_expectations.core import ExpectationSuite
from great_expectations.core.expectation_configuration import ExpectationConfiguration

# 假设json_schema包含期望配置
expectations_list = []
for ele in json_schema["expectations"]:
    config = ExpectationConfiguration(
        type=ele["expectation_type"],
        kwargs=ele["kwargs"]
    )
    expectations_list.append(config)

# 创建期望套件
suite = ExpectationSuite(
    name="my_suite",
    expectations=expectations_list
)
context.suites.add(suite)

总结

Great Expectations 1.0.5+ 版本已经解决了与 Spark Connect 的兼容性问题，用户现在可以无缝地在 Spark 3.5.0 及以上版本中使用该工具进行数据质量验证。对于从旧版本迁移的用户，建议仔细阅读官方文档，了解API变化，并按照新的模式重构验证流程。