DataHub 列级分类框架实战指南：配置、自研 Classifier 与源码级原理

DataHub 列级分类框架实战指南：配置、自研 Classifier 与源码级原理

【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub

DataHub 的分类（Classification）框架允许摄入源在 ingestion 过程中自动识别列的信息类型（info type），并将其作为业务术语（glossary term）挂载到对应字段上，从而在数据资产进入目录的第一时间完成敏感信息与业务语义的自动打标。这是一项**显式开启（opt-in）**的功能，默认关闭；内置的datahub分类器已被移除，但完整的框架——Classifier接口、分类器注册表与各数据源侧的编排逻辑——被完整保留，供开发者注册自己的分类器。读完本文，你将掌握分类配置的每个参数及其默认值、自研并注册一个Classifier的完整流程，以及从采样到术语落库的底层实现原理。

分类框架是什么

分类框架解决的是这样一个问题：当数据源（如 Snowflake、BigQuery、Redshift 等）的元数据被摄入 DataHub 时，框架会先读取每张表若干行的样本数据（sample values），把每一列的列名、描述、数据类型与样本值交给分类器（Classifier），由分类器预测该列属于哪种 info type（例如email、credit_card、phone_number等），最终将预测结果转换为对应的 glossary term 写入该列的SchemaMetadata。整个链路无需人工干预，敏感数据一进目录就被自动标注。

该功能在 YAML recipe 中以嵌套字段classification声明（文档明确规定用.表示嵌套层级），配置结构由 ClassificationConfig 定义，默认值为：enabled: false、sample_size: 100、max_workers: 进程 CPU 核数、table_pattern/column_pattern均放行全部、classifiers默认指向已移除的内置datahub类型。

重大变更：内置 datahub 分类器已被移除

仓库在 classifier_registry.py 的注释中明确记录了本次变更的背景：内置的datahub分类器（DataHubClassifier）依赖未维护的acryl-datahub-classify包，该包将numpy钉死在<2并携带一套过时的 spaCy 依赖栈，阻塞了整个摄入框架的依赖升级，因此被移除。随之下架的还有此前从该包导入的数据契约类型，现已在 classification_types.py 中原样内置（vendored），保证第三方分类器不需要该依赖即可使用扩展点。

移除之后，没有任何分类器在注册表中默认注册。如果 recipe 设置了classification.enabled: true却不注册替代分类器，框架会在启动阶段**快速失败（fail fast）**并给出指引。这条引导信息定义在 BUILTIN_CLASSIFIER_REMOVED_MESSAGE：

内置的datahub列级分类器已从 acryl-datahub 移除，因为它依赖未维护的acryl-datahub-classify包（钉死 numpy<2）。如需继续使用，请安装最后一个支持它的版本pip install 'acryl-datahub==1.6.0.5'；否则请注册你自己的分类器实现，或关闭分类功能（classification.enabled: false）。

该快速失败逻辑位于 classification_mixin.py：当从注册表查询分类器类型抛出KeyError时，若类型正是DEFAULT_CLASSIFIER_TYPE（即"datahub"），则抛出携带上述引导信息的ConfigurationError；若是其他未知类型，则抛出“在注册表中找不到该类型”的通用错误。注意get_classifiers()只在分类开启时才解析分类器——这样即使 recipe 保持默认的classifiers: [{type: datahub}]而未开启分类，任何数据源也不会误触发错误。

配置详解

以下表格完整列出分类功能的全部配置字段（字段与默认值均以当前仓库 classifier.py 与官方文档为准）：

完整 recipe 示例

以 Snowflake 源为例，一个开启分类并只针对特定表、特定列进行扫描的 recipe 如下：

source: type: snowflake config: # ... 数据源连接等基础配置 ... classification: enabled: true sample_size: 200 # 每列取 200 条样本值 max_workers: 4 # 4 个 worker 进程并行分类 info_type_to_term: email: "urn:li:glossaryTerm:customer.email" # 将 email info type 映射到自定义术语 table_pattern: allow: - "Customer.public.customer.*" deny: - "Customer.public.customer_staging.*" ignoreCase: true column_pattern: allow: - ".*\\.(email|phone|ssn)$" # 只分类这些后缀的列 deny: [] ignoreCase: true classifiers: - type: my-classifier

配置字段的源码级解读

• enabled 与 classifiers 的联动：ClassificationHandler.is_classification_enabled() 要求分类配置存在、enabled为真且classifiers非空，三者同时满足才算开启。
• sample_size 的 1.2 倍放大：框架不会只取恰好sample_size行。在 classification_workunit_processor 与 SQL 源的_classify方法中，实际请求的行数是sample_size * SAMPLE_SIZE_MULTIPLIER，其中SAMPLE_SIZE_MULTIPLIER = 1.2（见 classification_mixin.py）。原因写在 data_reader.py：跨列统一取样的查询难免混入 NULL，多取 20% 行可以补偿 NULL 对分类质量的影响。
• max_workers 与并行模型：当max_workers > 1时，框架使用concurrent.futures.ProcessPoolExecutor，以**每批 5 列（BATCH_SIZE）**为单位把列分发给多个 worker 进程并行调用分类器的classify（见 async_classify）。值得注意的实现细节：进程池显式使用multiprocessing.get_context("spawn")启动上下文——因为 Linux 上 Python 默认的fork启动方式在主进程使用线程时不安全（Python 3.14 起 Linux 默认也将切换为spawn）。如果自定义分类器包含无法序列化的状态，可把max_workers设为 1 走串行路径。
• 多分类器的优先级：文档与配置注释都强调“列表中靠后的分类器预测优先”。落到实现上，get_terms_for_column 对同一列的所有infotype_proposals取confidence_level最大者；而多个分类器按顺序逐个执行、update_field_terms以列为键覆盖写入，因此后执行的分类器（列表中靠后）的提案会覆盖先执行者的结果。
• info_type_to_term 的用途：分类器产出的是 info type（如email），而挂载到字段上的是 glossary term 的标识符。框架默认直接用 info type 字符串作为术语标识符；如需映射到既有术语，可通过info_type_to_term提供映射，未命中的 info type 仍回退为自身。
• pattern 的组合语义：table_pattern 与 column_pattern 是叠加过滤关系——列级判断is_classification_enabled_for_column同时要求表与列都匹配各自 pattern（见 classification_mixin.py）；它们还与父配置（如 Snowflake 的schema_pattern/table_pattern）组合使用，即“父级先过滤，分类 pattern 再过滤”。

分类的底层工作流

完整的分类链路分为四个阶段，可通过 ClassificationHandler 串联理解：

1. 构建分类器列表：get_classifiers()根据 recipe 中声明的类型逐一从classifier_registry解析并调用create(config_dict=...)实例化；未知类型抛出ConfigurationError（对内置datahub类型则抛出移除指引）。
2. 取样：通过DataReader.get_sample_data_for_table（抽象方法，见 data_reader.py）获取约sample_size * 1.2行数据，返回{列名: 值列表}字典。在 classify_schema_fields 中，若取样是惰性回调且执行失败，会累加num_tables_fetch_sample_values_failed统计并降级为空字典继续（同时提示确认数据集上的 SELECT 权限）。
3. 逐列构建 ColumnInfo 并调用分类器：get_columns_to_classify依据column_pattern过滤列，为每个字段构造ColumnInfo，其metadata携带Name、Description、DataType、Dataset_Name四项元数据（见 classification_types.py），values为样本值。随后串行或并行地交给各分类器，分类器回填infotype_proposals。
4. 写入 glossary terms：populate_terms_in_schema_metadata为每个命中列生成GlossaryTerms方面：通过make_term_urn(term)构造术语 URN，并保留字段上已有的术语（新术语排在前面，原术语追加在后），审计戳actor固定为datahub（见 classification_mixin.py）。

在 SQL 源侧，编排入口位于 sql_common.py 的SQLAlchemySource._classify方法——这是所有 SQLAlchemy 系 SQL 源的公共基类，它在生成 schema 元数据后调用ClassificationHandler.classify_schema_fields。非 SQLAlchemy 的源（如 Snowflake 的 snowflake_schema_gen.py、BigQuery 的 bigquery_schema_gen.py、Redshift 的 redshift.py）则通过classification_workunit_processor以 workunit 装饰器的方式接入，该处理器只拦截携带SchemaMetadata方面（MCP/MCPW）的 workunit，其余 workunit 原样透传。

自研分类器：Bring Your Own Classifier

这是当前框架唯一受支持的扩展路径。你需要做三件事：实现Classifier接口、注册到注册表、在 recipe 中引用。

1. 实现 Classifier 接口

Classifier是一个抽象基类（见 classifier.py），只要求实现classify(columns) -> columns一个抽象方法，并约定类方法create(config_dict)负责按配置实例化。输入输出都是ColumnInfo列表；分类器的职责是遍历输入的列、根据ColumnInfo.metadata（Name/Description/DataType/Dataset_Name）与ColumnInfo.values（样本值）做预测，并把预测结果填回column.infotype_proposals，然后原样返回列表。

文档给出的完整模板（可直接复制使用）：

from typing import Any, Dict, List from datahub.ingestion.glossary.classification_types import ColumnInfo from datahub.ingestion.glossary.classifier import Classifier from datahub.ingestion.glossary.classifier_registry import classifier_registry class MyClassifier(Classifier): def __init__(self, config: Dict[str, Any]) -> None: self.config = config @classmethod def create(cls, config_dict: Dict[str, Any]) -> "MyClassifier": return cls(config_dict or {}) def classify(self, columns: List[ColumnInfo]) -> List[ColumnInfo]: # Populate column.infotype_proposals here. return columns classifier_registry.register("my-classifier", MyClassifier)

2. 理解数据契约

ColumnInfo及其配套类型定义在 classification_types.py：

• ColumnInfo：metadata（Metadata对象）、values（样本值列表）、infotype_proposals（可空，预测结果）。
• Metadata：从meta_info字典派生出的name、description、datatype、dataset_name四个属性，取值键分别为Name、Description、Datatype、Dataset_Name；任一字段可能缺失（如列无描述），因此都是 Optional。
• InfotypeProposal：infotype（信息类型字符串）、confidence_level（置信度 float）、debug_info（各特征对预测的贡献占比，DebugInfo的name/description/datatype/values四个可选浮点字段）。

框架只消费InfotypeProposal.infotype与confidence_level：同列多个提案取置信度最高者（见 get_terms_for_column），并把命中列的完整名称（数据集名.列名）记入报告的info_types_detected。

3. 注册并启用

classifier_registry是PluginRegistry[Classifier]实例（见 classifier_registry.py），在模块导入时执行register("my-classifier", MyClassifier)即可完成注册。然后在 recipe 中引用：

source: type: snowflake config: # ... source config ... classification: enabled: true classifiers: - type: my-classifier

若分类器需要自定义参数，可通过classifiers[].config传入任意 JSON/YAML 对象，框架会原样交给create(config_dict=...)；config字段的完整校验由你的分类器自己负责（见 classifier.py 中DynamicTypedClassifierConfig的注释）。

4. 常见排查要点

• 启动即报The built-in 'datahub' column-level classifier has been removed...：说明 recipe 开了分类但没注册任何分类器，且classifiers仍是默认值。按引导信息二选一：注册自己的分类器，或关闭classification.enabled。
• 报Cannot find classifier class of type=xxx in the registry!：classifiers里写的类型未注册，检查拼写与注册语句是否被执行。
• 某张表没被分类：检查table_pattern/column_pattern与父级过滤是否放行；检查分类是否开启且classifiers非空（见is_classification_enabled_for_table的三重条件，classification_mixin.py）。
• 取样失败：报告中的num_tables_fetch_sample_values_failed会增加，日志提示确认对数据集授予了 SELECT 权限（classification_mixin.py）。
• 并行模式异常：ProcessPoolExecutor使用spawn上下文，分类器对象需可序列化；否则将max_workers设为 1 降级为进程内串行。

支持的源

文档明确声明：所有 SQL 源均支持分类。从仓库代码看，支持方式分两类：

• 基于 SQLAlchemy 的通用 SQL 源：继承 SQLAlchemySource（含 Oracle、Teradata、HANA、Doris 等具体实现），在公共基类中完成编排；
• 各自实现采样与编排的专用源：Snowflake（snowflake_schema_gen.py）、BigQuery（bigquery_schema_gen.py）、Redshift（redshift.py）等通过ClassificationSourceConfigMixin混入分类配置（见各自 config 文件，如 snowflake_config.py、bigquery_config.py），再经由classification_workunit_processor挂接处理。

此外，非 SQL 的 DynamoDB 源（dynamodb.py）同样接入了该框架，说明凡是能提供样本数据的源都可以复用这套机制。可以推断，只要你的自定义源混入ClassificationSourceConfigMixin并接入ClassificationHandler或classification_workunit_processor，就能获得与内置 SQL 源一致的分级能力。

小结

DataHub 分类框架的价值在于把“数据进入目录”与“语义自动标注”绑定为一次性的流水线动作。当前版本的关键事实是：内置分类器已随acryl-datahub-classify依赖的移除而下线（如需保留请锁定acryl-datahub==1.6.0.5），但接口、注册表与编排层完整保留且文档与代码高度自洽——你只需实现一个classify()方法并在注册表中登记，即可在全部 SQL 源上获得配置灵活（采样量、并行度、表列过滤、术语映射、多分类器优先级）的自动术语识别能力。延伸阅读可参考 classification 官方文档、分类配置模型、分类编排实现 与 数据契约类型。

【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub