在Airflow本地Docker环境测试Great Expectations数据源连接时出现KeyError：找不到module_name键

我是Great Expectations（GX）的新手，最近在本地Docker环境的Airflow里尝试测试great_expectations.yml中定义的数据源连接，写了下面的测试函数，但运行时碰到了KeyError，提示找不到module_name键。

我的测试函数代码： #

import logging
import yaml
import great_expectations as ge

def test_ge_data_context_connections():
    context = ge.data_context.DataContext()

    # Test each datasource
    for datasource in context.list_datasources():
        datasource_name = datasource['name']
        logging.info(f"Testing connection for datasource: {datasource_name}...")
        try:
            datasource_config = context.get_datasource(datasource_name)
            datasource_yaml = yaml.dump(datasource_config.config)

            connection_test_result = context.test_yaml_config(datasource_yaml)
            # 后续处理逻辑因报错中断
        except Exception as e:
            logging.error(f"Failed to test {datasource_name}: {str(e)}")
            raise

报错信息片段： #

File "/home/airflow/.local/lib/python3.11/site-packages/great_expectations/data_context/config_validator/yaml_config_validator.py", line 556, in _test_instantiation_of_misc_class_from_yaml_config
    instantiated_class = instantiate_class_from_config(
KeyError: 'module_name'

问题分析： #

这个错误的核心原因是test_yaml_config()方法需要完整的、符合GX规范的顶层YAML配置结构，但你通过datasource_config.config导出的内容，只是数据源对象的内部子配置，缺少了GX实例化数据源时必须的module_name和class_name这些核心标识字段。

举个例子，GX要求的数据源完整配置YAML应该是这样的：

datasources:
  my_postgres_ds:
    class_name: Datasource
    module_name: great_expectations.datasource
    execution_engine:
      class_name: SqlAlchemyExecutionEngine
      module_name: great_expectations.execution_engine
      connection_string: postgresql://user:pass@host:port/db
    data_connectors:
      default_inferred_data_connector_name:
        class_name: InferredAssetSqlDataConnector
        module_name: great_expectations.datasource.data_connector
        include_schema_name: true

而你当前导出的内容可能只包含了execution_engine、data_connectors这类子节点，丢失了顶层的标识字段，导致GX无法识别要实例化的数据源类。

解决办法： #

方法1：直接使用完整的原始配置

从DataContext的原始配置中读取完整的数据源定义，避免丢失关键字段：

def test_ge_data_context_connections():
    context = ge.data_context.DataContext()
    # 获取great_expectations.yml中所有数据源的完整配置
    all_datasources = context.config.get("datasources", {})

    for datasource_name, datasource_full_config in all_datasources.items():
        logging.info(f"Testing connection for datasource: {datasource_name}...")
        try:
            # 构建包含顶层datasources键的完整YAML结构
            full_config_yaml = yaml.dump({
                "datasources": {
                    datasource_name: datasource_full_config
                }
            })
            # 执行配置测试
            test_result = context.test_yaml_config(full_config_yaml)
            
            if test_result.success:
                logging.info(f"✅ Connection test passed for {datasource_name}")
            else:
                logging.error(f"❌ Connection test failed for {datasource_name}: {test_result.message}")
        except Exception as e:
            logging.error(f"⚠️ Error testing {datasource_name}: {str(e)}")

方法2：手动补充缺失的标识字段

如果坚持要用get_datasource()获取的对象，可以手动补充module_name和class_name字段：

def test_ge_data_context_connections():
    context = ge.data_context.DataContext()

    for datasource in context.list_datasources():
        datasource_name = datasource['name']
        logging.info(f"Testing connection for datasource: {datasource_name}...")
        try:
            datasource_obj = context.get_datasource(datasource_name)
            # 构建包含核心标识字段的完整配置
            full_config = {
                "class_name": datasource_obj.__class__.__name__,
                "module_name": datasource_obj.__module__,
                **datasource_obj.config
            }
            # 包装成GX要求的顶层结构并转YAML
            full_yaml = yaml.dump({
                "datasources": {
                    datasource_name: full_config
                }
            })
            test_result = context.test_yaml_config(full_yaml)
            # 处理测试结果...
        except Exception as e:
            logging.error(f"Error testing {datasource_name}: {str(e)}")

额外检查项：

• 确保Airflow Docker容器中的GX版本和你本地开发环境版本一致，不同版本的GX配置结构可能存在差异，版本不匹配也可能引发这类问题。可以在容器内执行pip show great_expectations查看版本。
• 检查你的great_expectations.yml中每个数据源是否都正确配置了module_name和class_name字段，避免配置本身就有缺失。

备注：内容来源于stack exchange，提问作者Ana