📚 课程目录第4章连接数据源

4、连接数据源：配置 Pandas、Spark、SQL 数据库等数据源，创建 Data Connector

数据校验的第一步是什么？

不是写规则，不是调参数。是把数据先「接进来」。

Great Expectations 支持三种主流数据后端：Pandas、Spark 和 SQL 数据库。每种后端都有自己的脾气，配置方式也略有不同。这一章，我带你逐个搞定它们。

4.1 数据源与 Data Connector 的关系

先理清两个概念：Datasource 和 Data Connector。

Datasource：数据源的抽象。它告诉 GE 你要连什么类型的数据（Pandas DataFrame？Spark 表？还是 MySQL 数据库？）。
Data Connector：数据连接器。它负责「找到」具体的数据文件或表，并生成 BatchRequest。

说白了，Datasource 是「插座」，Data Connector 是「插头」。你先把插座装好，再决定插头怎么接。

核心要点：一个 Datasource 可以挂多个 Data Connector。比如同一个 SQL 数据库，你可以用 ConfiguredAssetSqlDataConnector 精确指定表名，也可以用 InferredAssetSqlDataConnector 自动扫描所有表。

下面这张图能帮你快速理解整体架构：

4.2 配置 Pandas Datasource

Pandas 是最轻量的数据后端。适合单机、小规模数据、快速原型验证。

我个人习惯在开发阶段先用 Pandas，等逻辑稳定了再切到 Spark 或 SQL。

4.2.1 基本配置

import great_expectations as gx

context = gx.get_context()

# 创建 Pandas Datasource
datasource = context.sources.add_pandas(
    name="my_pandas_datasource"
)

print(datasource)
# 输出: <PandasDatasource "my_pandas_datasource">

就这么简单？嗯，对 Pandas 来说确实简单。因为 Pandas 不需要连接信息，它直接操作内存中的 DataFrame。

4.2.2 添加 Data Connector

Pandas Datasource 默认自带一个 InferredAssetDataConnector。但如果你想精确控制文件路径，可以手动添加：

# 添加 ConfiguredAsset Data Connector
datasource.add_configured_data_connector(
    name="my_connector",
    assets={
        "orders": {
            "base_directory": "./data/orders",
            "pattern": "orders_(.*)\\.csv"
        },
        "customers": {
            "base_directory": "./data/customers",
            "pattern": "customers_(.*)\\.parquet"
        }
    }
)

小技巧：用 pattern 参数可以提取文件名中的变量。比如 orders_202401.csv 会提取出 202401 作为 batch 标识。我在做月度数据校验时经常用这个。

4.3 配置 Spark Datasource

数据量大了怎么办？上 Spark。

Spark Datasource 的配置比 Pandas 多一步：你需要提供 SparkSession。

4.3.1 基本配置

from pyspark.sql import SparkSession

# 创建 SparkSession
spark = SparkSession.builder \
    .appName("GE_Spark") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# 创建 Spark Datasource
datasource = context.sources.add_spark(
    name="my_spark_datasource",
    spark_session=spark
)

注意：spark_session 参数必须传。不传的话 GE 会尝试自己创建一个，但那个 Session 可能没有你需要的配置。

我曾经踩过的坑：在 EMR 集群上跑 GE 时，忘了传 SparkSession，结果 GE 自己创建的 Session 没有启用动态资源分配，导致大表校验时 OOM。后来我强制传入生产环境的 SparkSession，问题解决。

4.3.2 读取 Hive 表

Spark 最常见的场景是读 Hive 表。配置方式如下：

datasource.add_configured_data_connector(
    name="hive_connector",
    assets={
        "sales": {
            "table_name": "default.sales"
        },
        "inventory": {
            "table_name": "default.inventory"
        }
    }
)

你想想看，这样配置之后，每次校验只需要指定 asset 名称，GE 会自动去 Hive 元数据里找对应的表。

4.4 配置 SQL Datasource

SQL 数据库是企业级数据校验的主力。GE 支持 MySQL、PostgreSQL、BigQuery、Snowflake 等主流数据库。

4.4.1 基本配置

datasource = context.sources.add_sql(
    name="my_sql_datasource",
    connection_string="postgresql://user:password@host:5432/mydb"
)

连接字符串的格式遵循 SQLAlchemy 规范。常用的几种：

数据库	连接字符串格式
PostgreSQL	`postgresql://user:pass@host:5432/db`
MySQL	`mysql+pymysql://user:pass@host:3306/db`
BigQuery	`bigquery://project-id/dataset`
Snowflake	`snowflake://user:pass@account/db/schema`

4.4.2 两种 Data Connector 模式

SQL Datasource 支持两种 Data Connector：

InferredAssetDataConnector：自动扫描数据库中的所有表。适合探索阶段。
ConfiguredAssetDataConnector：手动指定要连接的表。适合生产环境。

看个对比：

# 自动扫描（Inferred）
datasource.add_inferred_data_connector(
    name="all_tables"
)
# 这样配置后，GE 会读取数据库中所有用户表

# 手动指定（Configured）
datasource.add_configured_data_connector(
    name="critical_tables",
    assets={
        "orders": {"table_name": "orders"},
        "payments": {"table_name": "payments"}
    }
)
# 只连接 orders 和 payments 两张表

我的建议：生产环境用 ConfiguredAsset。自动扫描虽然方便，但万一有人新建了一张测试表，你的校验任务可能会报错。手动指定更可控。

4.5 创建 BatchRequest

配置好 Datasource 和 Data Connector 之后，怎么真正拿到数据？

答案是 BatchRequest。

batch_request = datasource.get_batch_request(
    data_connector_name="critical_tables",
    data_asset_name="orders"
)

# 也可以指定批次
batch_request = datasource.get_batch_request(
    data_connector_name="my_connector",
    data_asset_name="orders",
    batch_identifiers={"month": "202401"}
)

BatchRequest 就像一张「取货单」。它告诉 GE：我要从哪个 Connector、哪个 Asset、哪个批次拿数据。

4.6 实战：完整配置示例

最后，我给你一个完整的配置流程。假设我们要连接一个 PostgreSQL 数据库，只校验 orders 和 customers 两张表：

import great_expectations as gx

# 1. 获取 Context
context = gx.get_context()

# 2. 创建 SQL Datasource
datasource = context.sources.add_sql(
    name="production_db",
    connection_string="postgresql://admin:pass@prod-host:5432/analytics"
)

# 3. 添加 Configured Data Connector
datasource.add_configured_data_connector(
    name="core_tables",
    assets={
        "orders": {"table_name": "orders"},
        "customers": {"table_name": "customers"}
    }
)

# 4. 创建 BatchRequest
orders_request = datasource.get_batch_request(
    data_connector_name="core_tables",
    data_asset_name="orders"
)

# 5. 开始校验
validator = context.get_validator(
    batch_request=orders_request
)

# 执行一个简单的 Expectation
validator.expect_column_values_to_not_be_null("order_id")
validator.save_expectation_suite("orders_suite")

你看，从配置到校验，一共就 5 步。结构清晰，逻辑明确。

补充一点：如果你用的是 Pandas Datasource，第 2 步换成 add_pandas()，第 3 步的 table_name 换成 base_directory 和 pattern。其他步骤完全一样。

数据源配置这块，说白了就是「对号入座」。Pandas 轻量、Spark 分布式、SQL 企业级。选对后端，配好 Connector，你的数据校验管道就搭好了地基。