Databricks中使用Great_Expectations遇SparkConf构造函数未白名单错误

环境信息 #

• Databricks共享集群，运行时版本13.1 Beta（包含Apache Spark 3.4.0、Scala 2.12）
• 依赖版本：py4j 0.10.9.7、pyspark 3.4.0

执行代码 #

%pip install great_expectations
dbutils.library.restartPython()

import great_expectations as gx
from great_expectations.checkpoint import SimpleCheckpoint

context_root_dir = "abfss://<container>@<acc>.dfs.core.windows.net/tmp/great_expectations/"
context = gx.get_context(context_root_dir=context_root_dir)
print(context)

from pyspark.sql import SparkSession
import pandas as pd

session_name = 'mk_spark_session'
spark = SparkSession.builder.appName(session_name).getOrCreate()

query = "SELECT * FROM my_test_table limit 10"
spark_df = spark.sql(query)

dataframe_datasource = context.sources.add_or_update_spark(
    name="my_spark_in_memory_datasource",
)
print(dataframe_datasource)

dataframe_asset = dataframe_datasource.add_dataframe_asset(
    name="MK_DF_asset",
    dataframe=spark_df,
)
print(dataframe_asset)

# 重复添加数据源和资产（无必要操作）
dataframe_datasource = context.sources.add_or_update_spark(
    name="my_spark_in_memory_datasource",
)
print(dataframe_datasource)

dataframe_asset = dataframe_datasource.add_dataframe_asset(
    name="MK_DF_asset",
    dataframe=spark_df,
)
print(dataframe_asset)

batch_request = dataframe_asset.build_batch_request()
print(batch_request)

# 创建期望套件
expectation_suite_name = "MK_expectation_suite"
context.add_or_update_expectation_suite(expectation_suite_name=expectation_suite_name)

# 报错代码行
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name=expectation_suite_name,
)

print(validator.head())

报错信息 #

py4j.security.Py4JSecurityException: Constructor public org.apache.spark.SparkConf(boolean) is not whitelisted.

Py4JError: An error occurred while calling None.org.apache.spark.SparkConf. Trace:
py4j.security.Py4JSecurityException: Constructor public org.apache.spark.SparkConf(boolean) is not whitelisted.
  at py4j.security.WhitelistingPy4JSecurityManager.checkConstructor(WhitelistingPy4JSecurityManager.java:451)
  at py4j.Gateway.invoke(Gateway.java:256)
  at py4j.commands.ConstructorCommand.invokeConstructor(ConstructorCommand.java:80)
  at py4j.commands.ConstructorCommand.execute(ConstructorCommand.java:69)
  at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:195)
  at py4j.ClientServerConnection.run(ClientServerConnection.java:115)
  at java.lang.Thread.run(Thread.java:750)

原因分析 #

错误源于Databricks共享集群的安全白名单限制：Great Expectations在初始化时尝试调用org.apache.spark.SparkConf(boolean)构造方法，而该方法未被Databricks的安全策略允许。同时代码中手动创建SparkSession的操作会与Databricks内置的Spark实例产生冲突，加剧了问题。

解决方案 #

1. 使用Databricks适配模式初始化GX Context #

直接通过mode="databricks"参数初始化GX Context，自动适配Databricks环境的安全限制：

%pip install great_expectations
dbutils.library.restartPython()

import great_expectations as gx
from great_expectations.checkpoint import SimpleCheckpoint

# 适配Databricks环境的Context初始化方式
context = gx.get_context(mode="databricks")
print(context)

# 直接使用Databricks内置的spark对象，无需手动创建
query = "SELECT * FROM my_test_table limit 10"
spark_df = spark.sql(query)

# 添加Spark数据源时传入内置的spark实例
dataframe_datasource = context.sources.add_or_update_spark(
    name="my_spark_in_memory_datasource",
    spark=spark
)
print(dataframe_datasource)

dataframe_asset = dataframe_datasource.add_dataframe_asset(
    name="MK_DF_asset",
    dataframe=spark_df,
    # 可选：自定义元数据
    batch_metadata={"source_table": "my_test_table", "row_limit": 10}
)
print(dataframe_asset)

batch_request = dataframe_asset.build_batch_request()
print(batch_request)

expectation_suite_name = "MK_expectation_suite"
context.add_or_update_expectation_suite(expectation_suite_name=expectation_suite_name)

# 正常获取Validator
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name=expectation_suite_name,
)

print(validator.head())

2. 关于batch_metadata为空的说明 #

空batch_metadata是正常现象，因为你直接传入内存中的DataFrame，没有关联外部数据源的元数据（如文件路径、分区信息）。如果需要添加自定义元数据，可参考上述代码中的batch_metadata参数设置。

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