数据源接入与探查：先摸清家底，再谈治理
数据质量工程的第一步，不是写规则，也不是搭平台。而是——你得先知道你的数据长什么样、从哪来、能信几分。
我见过太多团队，一上来就搞数据质量规则，结果发现源系统字段类型都变了，或者某些表根本没人维护。说白了，数据源接入和探查，就是给数据做一次「入职体检」。这一步做扎实了，后面的治理工作才能站得住脚。
数据源类型识别：你的数据是哪种「物种」？
数据源不是铁板一块。我个人习惯，先把数据源分成三大类：
| 类型 | 典型代表 | 接入特点 | 质量风险 |
|---|---|---|---|
| 结构化 | MySQL、PostgreSQL、Oracle、SQL Server | Schema 固定，JDBC/ODBC 直连 | 字段类型变更、空值率波动 |
| 半结构化 | JSON 日志、XML、CSV、Parquet | Schema on read，需解析器 | 字段缺失、嵌套层级不一致 |
| 非结构化 | 图片、PDF、音视频、文本 | 对象存储或消息队列 | 元数据缺失、内容无法直接校验 |
你想想看，一个 JSON 日志文件，今天多了一个字段，明天少了一个层级，如果你用固定 Schema 去解析，直接就崩了。所以识别数据源类型，决定了你用什么工具、什么策略去探查。
数据探查工具：Great Expectations 实战
说到数据探查工具，Great Expectations（简称 GX）是我目前用得最顺手的。为什么？因为它把「探查」和「断言」合在了一起。你探查出来的结果，直接就能转成质量规则。
我记得第一次在项目里用 GX，团队里还有人质疑：「这不就是个测试框架吗？」嗯，还真不是。GX 的核心是 Expectation——你对数据的一个预期。比如：
expect_column_values_to_not_be_null("user_id")—— user_id 不能为空expect_column_values_to_be_between("age", 0, 150)—— 年龄在合理范围expect_table_row_count_to_be_between(1000, 100000)—— 行数不能太少或太多
下面是一个典型的 GX 探查流程：
# 1. 安装
pip install great_expectations
# 2. 初始化数据上下文
import great_expectations as gx
context = gx.get_context()
# 3. 连接数据源
datasource = context.sources.add_pandas("my_datasource")
data_asset = datasource.add_dataframe_asset("my_asset")
batch_request = data_asset.build_batch_request(dataframe=df)
# 4. 创建 Expectation Suite
suite = context.add_expectation_suite("user_profile_suite")
# 5. 添加探查期望
suite.add_expectation(
gx.expectations.ExpectColumnValuesToNotBeNull(column="user_id")
)
suite.add_expectation(
gx.expectations.ExpectColumnValuesToBeBetween(column="age", min_value=0, max_value=150)
)
# 6. 运行验证
validator = context.get_validator(
batch_request=batch_request,
expectation_suite=suite
)
results = validator.validate()
print(results["statistics"])
跑完之后，GX 会生成一个 HTML 报告，哪些字段通过了、哪些失败了、失败率多少，一目了然。我曾经在一个电商项目里，用 GX 跑了一遍订单表，发现「支付金额」字段有 3% 的记录是负数——业务方自己都不知道。这就是探查的价值。
元数据采集：给数据做「身份证」
元数据是什么？说白了，就是「关于数据的数据」。字段名、类型、长度、注释、主键、外键、血缘关系……这些都是元数据。
没有元数据，你连这个字段是干嘛的都不知道，谈何质量治理？
我常用的元数据采集方式有三种：
- JDBC 元数据接口：通过
DatabaseMetaData获取表结构、主键、索引。适合关系型数据库。 - 文件头解析：Parquet 的 schema、CSV 的第一行、JSON 的 key 集合。适合半结构化数据。
- 日志/API 采集：从数据源的管理接口或审计日志中获取变更记录。适合实时监控。
举个例子，用 Python 采集 MySQL 元数据：
import pymysql
conn = pymysql.connect(host="localhost", user="root", password="pass", database="test")
cursor = conn.cursor()
# 获取所有表
cursor.execute("SHOW TABLES")
tables = cursor.fetchall()
for table in tables:
table_name = table[0]
# 获取表注释
cursor.execute(f"SHOW TABLE STATUS WHERE Name = '{table_name}'")
status = cursor.fetchone()
comment = status[17] # 第18列是注释
print(f"表: {table_name}, 注释: {comment}")
# 获取字段信息
cursor.execute(f"SHOW FULL COLUMNS FROM {table_name}")
columns = cursor.fetchall()
for col in columns:
print(f" 字段: {col[0]}, 类型: {col[1]}, 注释: {col[8]}")
知识体系总览
下面这张图，把数据源接入与探查的核心逻辑串起来了。你可以把它当作本章的「思维导图」：
接入与探查的常见陷阱
做了这么多年数据治理，我踩过不少坑。挑几个典型的说说：
- 陷阱一：只探查一次。数据源是会变的。字段类型从 int 变成 varchar，你第一次探查没问题，三个月后可能就全崩了。我建议至少每周跑一次全量探查，每天跑一次增量探查。
- 陷阱二：忽略空值率。很多团队只看字段有没有值，不看空值率的变化。比如「手机号」字段，昨天空值率 2%，今天突然变成 20%——这往往是上游接口出了问题。
- 陷阱三：元数据采集太「重」。我曾经见过一个团队，用 Apache Atlas 做全量元数据采集，部署了 10 台服务器，结果每天采集的数据量还没元数据本身大。小团队的话，用 GX 自带的元数据功能，或者写个轻量脚本就够了。