大数据智能体工作台 DataBuddy 特征工程 SDK

一、前提条件

·已开通 DataBuddy 服务并完成空间初始化。

·当前账号在目标 Catalog / Schema 上拥有相应的数据读写权限。

·已创建并启用一个可用的计算资源（创建方法详见 创建计算资源）。

·运行特征工程代码的内核环境 Python 版本 ≥ 3.10 。

·已在 Notebook / Python 内核中安装特征工程 SDK（要求 0.2.3 及以上 ）：

%pip install "wedata3-feature-engineering[mlflow3]>=0.2.3"

注意：这篇示例基于 0.2.3+ 版本验证。SDK 依赖 mlflow >= 3.10.0, < 3.11.0，若 import wedata 时提示 mlflow 版本不满足，请确认安装的是 [mlflow3] extra（即 wedata3-feature-engineering[mlflow3]），且不要在同一 Notebook 内混装 mlflow 2.x 与 3.x。

注意：在 DataBuddy Notebook 内核中运行时，鉴权所需环境变量会被自动注入，无需手动配置；若在本地 IDE 调试，则需自行 export 环境变量，否则初始化 FeatureEngineeringClient(…) 时会抛 EnvironmentError / KeyError。

二、初始化客户端

在代码单元中导入并初始化特征工程客户端。DataBuddy Notebook 内核已预置 spark 会话，直接传入即可；本地调试时需自行创建 SparkSession 并传入。

%pip install "wedata3-feature-engineering[mlflow3]>=0.2.3"%pip list | grep wedata3-feature-engineering

from wedata.feature_engineering.client import FeatureEngineeringClientfrom wedata.common.constants.engine_types import EngineTypes# DataBuddy Notebook 内核已预置 spark 会话；本地调试时需自行创建并传入fe = FeatureEngineeringClient(spark)

三、特征数据库管理

特征表归属于某个 Schema（特征数据库）。若目标 Schema 尚未创建，可先通过 SDK 幂等准备；若已规划好 Schema，可跳过本节直接建表。

# 幂等创建特征数据库（已存在则跳过）；支持单段名或二段式 catalog.schemafe.create_database("<schema_name>", catalog_name="<catalog_name>", comment="<数据库描述>")# 列出指定 Catalog 下的全部数据库fe.list_databases(catalog_name="<catalog_name>")# 判断数据库是否存在（任何内部异常都会被吞掉并返回 False）fe.database_exists("<schema_name>", catalog_name="<catalog_name>")# 删除数据库，默认非级联以避免误删；级联删除请慎用fe.drop_database("<schema_name>", catalog_name="<catalog_name>")

注意：drop_database 在数据库非空且 cascade=False 时会由底层抛错；设置 cascade=True 会级联删除其下所有表，生产环境请谨慎使用。

四、创建特征表

调用 create_table 创建特征表。表结构可由传入的 df 自动推断，也可显式传入 schema；engine_type 为必填项，用于指定底层表引擎：

from wedata.common.constants.engine_types import EngineTypes# 方式一：传入 df，自动推断 schema 并写入首批数据fe.create_table( name="<catalog>.<schema>.<table_name>", # 特征表全限定名 primary_keys=["<主键列表>"], # 单键传字符串，复合主键传列表 timestamp_key=["<时间戳列名>"], # 时间戳键列名（单数） df=<Spark DataFrame>, # 传入 df 可自动推断 schema 并写入首批数据 engine_type=EngineTypes.ICEBERG_ENGINE, # 必填：ICEBERG_ENGINE 或 HIVE_ENGINE description=["<表描述>"], # 可选)# 方式二：不传 df，显式提供 schema 建空表fe.create_table( name="<catalog>.<schema>.<table_name>", primary_keys=["<主键列名>"], timestamp_key="<时间戳列名>", schema=<PySpark StructType>, # df为空时必填 engine_type=EngineTypes.ICEBERG_ENGINE,)

注意：1. primary_keys 类型为 Union[str, List[str]]，单主键可直接传字符串，复合主键传列表。

注意：2. schema 与 df 二选一：当 df 已提供时 schema 可省略（由 df 推断）；仅当 df=None 时 schema 必填。

注意：3. 其他可选参数：catalog_name / database_name（在 name 未使用全限定名时指定）、tags（表级标签字典）、partition_columns（分区列）。

注意：引擎选择（engine_type）：EngineTypes 提供 ICEBERG_ENGINE（Iceberg 表，推荐）与 HIVE_ENGINE（Hive 表）两种取值。优先选用 Iceberg：它支持 schema evolution 与 time travel，在按时间点回溯特征查询时性能更优。

注意：写入前的数据类型治理 ：特征工程 SDK 不识别 decimal(p, s) 类型——它会导致记录模型时无法推断模型签名（signature），进而引发在线、离线推理的类型不匹配。建议在写入特征表之前，把所有 decimal 列统一 cast("double")，并将该处理沉淀为可复用的工具函数。时间戳列应保留为时间类型，不要转为字符串，否则无法用作时间戳键。

执行成功后，新建的特征表将自动出现在特征管理列表中。

注意：主键与时间戳的选择建议 ：主键用于唯一定位一条特征记录；时间戳键用于实现按时间点回溯特征值。建议将业务实体的唯一标识作为主键、特征生成时间作为时间戳键。

五、写入特征表

如需追加写入更多特征数据，调用 write_table：

fe.write_table( name="<catalog>.<schema>.<table_name>", # 目标特征表全限定名 df=<Spark DataFrame>, # 待写入的特征数据 mode="append", # 写入模式：append（默认）/ overwrite)

流式写入时需额外提供 checkpoint 路径：

fe.write_table( name="<catalog>.<schema>.<table_name>", df=<Streaming DataFrame>, mode="append", checkpoint_location="<checkpoint 路径>", # 流式写入必填 trigger={"processingTime": "10 seconds"}, # 可选，流式触发器配置)

写入模式说明 ：

·append ：追加写入，新数据按主键合并入表（默认模式）。

·overwrite ：覆盖整张表的全部数据，慎用于生产场景。

注意：当前 SDK（0.2.4）write_table 不支持 merge 模式。增量写入请使用默认的 append。

六、获取与读取特征表

调用 get_table 可以编程方式拿到特征表的完整元数据；调用 read_table 读取特征表数据（返回 Spark DataFrame）：

# 获取元数据table_info = fe.get_table(name="<catalog_name>.<schema_name>.<table_name>")print(table_info.primary_keys)print(table_info.timestamp_keys)print(table_info.tags)# 读取数据df = fe.read_table(name="<catalog_name>.<schema_name>.<table_name>")df.show(5)

注意：read_table 读取的是离线特征表中的全量数据。在线特征表（PostgreSQL）中的数据用于在线推理服务在收到请求时自动查找特征，无需调用方在 SDK 中显式读取；如需核对在线表数据，可在数据目录中查看对应的在线特征表。

七、维护特征表标签

特征表标签可用于版本号、所属业务线、特征类别等业务分类，可通过 SDK 管理：

# 设置或更新一个标签fe.set_feature_table_tag( name="<catalog_name>.<schema_name>.<table_name>", key="<tag_key>", # 如 "business_line"、"version" 等 value="<tag_value>", # 如 "growth"、"v1.2.0" 等)# 删除一个标签fe.delete_feature_table_tag( name="<catalog_name>.<schema_name>.<table_name>", key="<tag_key>",)

设置或删除完成后，标签会同步显示在 Catalog 详情页与特征管理列表的 标签 列。推荐为每张特征表至少维护以下标签，便于检索与治理：

注意：primary_keys、timestamp_keys 等是 SDK 的保留字段，不能 作为自定义标签的 key 使用。

注意：版本管理建议 ：DataBuddy 当前未提供独立的特征表版本号对象，建议使用 version 标签（如 v1.0.0、v1.1.0）配合 Catalog 表本身的数据变更历史来标识业务版本演进。如需在不同版本之间快速切换，可采用按版本建独立特征表、在训练代码中显式选择的方式。

八、删除特征表

当一张特征表不再被任何模型与下游使用时，可通过 SDK 删除：

fe.drop_table(name="<catalog_name>.<schema_name>.<table_name>")

注意：警告

注意：删除特征表会同时删除底层数据文件，该操作不可恢复。删除前请确认：

九、构建训练数据集

以一组主键（如实体 ID、事件时间戳）作为 DataFrame，调用 create_training_set 拼接多张特征表生成训练数据：

from wedata.feature_engineering.client import FeatureEngineeringClientfrom wedata.common.entities.feature_lookup import FeatureLookupfe = FeatureEngineeringClient(spark)# 1. 主键 DataFrame：包含训练标签 (label) + 主键 + 事件时间戳labels_df = spark.table("<catalog_name>.<schema_name>.<label_table_name>")# 2. 定义需要拼接的特征表与字段（强烈推荐使用三段式全限定名，避免依赖默认 catalog）feature_lookups = [ FeatureLookup( table_name="<catalog_name>.<schema_name>.<feature_table_name_1>", feature_names=["<feature_column_1>", "<feature_column_2>"], lookup_key=["<primary_key>"], timestamp_lookup_key="<timestamp_key>", ), FeatureLookup( table_name="<catalog_name>.<schema_name>.<feature_table_name_2>", feature_names=["<feature_column_3>"], lookup_key=["<primary_key>"], timestamp_lookup_key="<timestamp_key>", ),]# 3. 创建训练数据集training_set = fe.create_training_set( df=labels_df, feature_lookups=feature_lookups, label="<label_column>", exclude_columns=[],)training_df = training_set.load_df()

注意：timestamp_lookup_key 用于按时间点回溯特征值（point-in-time correctness），即对每条样本拼接其事件时间点之前最近一次的特征值，避免特征穿越。如需限制回溯的时间范围，可在 FeatureLookup 中配合 lookback_window（datetime.timedelta）使用，仅查找该时间窗内的特征值。

注意：数据类型治理 ：写入与拼接前务必将 decimal(p,s) 转为 double。SDK 不识别 decimal(p,s) 类型，否则可能导致类型解析失败或在线、离线值不一致。

十、回溯窗口与多表 / On-Demand 特征组合

1. 回溯窗口 ：lookback_window 必须与 timestamp_lookup_key 同时使用，且取值为非负 的 datetime.timedelta（0 表示精确匹配，不传表示不限制），否则会抛 ValueError。

2. 多 FeatureLookup + FeatureFunction 组合 ：可同时声明多张特征表查找，并通过 FeatureFunction 引入按需计算（On-Demand）特征：

import datetimefrom wedata.common.entities.feature_lookup import FeatureLookupfrom wedata.common.entities.feature_function import FeatureFunctionfeature_lookups = [ FeatureLookup( table_name="<catalog_name>.<schema_name>.<feature_table_a>", lookup_key=["<primary_key_a>"], feature_names=["<feature_col_1>", "<feature_col_2>"], rename_outputs={"<feature_col_1>": "<renamed_col_1>"}, # 输出列重命名，避免同名冲突 ), FeatureLookup( table_name="<catalog_name>.<schema_name>.<feature_table_b>", lookup_key=["<primary_key_b>"], timestamp_lookup_key="<timestamp_key>", lookback_window=datetime.timedelta(days=30), # 仅取近 30 天内的特征值 ), FeatureFunction( # 按需计算特征（On-Demand UDF） udf_name="<catalog_name>.<schema_name>.<udf_name>", input_bindings={"<udf_input_param>": "<renamed_col_1>"}, output_name="<udf_output_col>", ),]

十一、训练并记录模型

使用拼接好的训练 DataFrame 训练模型，并通过 log_model 把模型连同特征来源信息一起记录到 MLflow：

import mlflowfrom sklearn.pipeline import Pipelinefrom sklearn.preprocessing import OneHotEncoder, StandardScalerfrom sklearn.compose import ColumnTransformerfrom sklearn.ensemble import GradientBoostingClassifierwith mlflow.start_run() as run: pdf = training_df.toPandas() X = pdf.drop(columns=["<label_column>"]) y = pdf["<label_column>"] # 预处理（OHE / Scaler 等）必须封装进 Pipeline，禁止在 Pipeline 外部用pd.get_dummies 等手工预处理， # 否则在线推理时无法复现训练期的特征变换，导致线上线下不一致。 preprocessor = ColumnTransformer( transformers=[ ("cat", OneHotEncoder(handle_unknown="ignore"), ["<categorical_feature>"]), ("num", StandardScaler(), ["<numeric_feature_1>", "<numeric_feature_2>"]), ] ) model = Pipeline(steps=[ ("preprocessor", preprocessor), ("clf", GradientBoostingClassifier()), ]) model.fit(X, y) fe.log_model( model=model, artifact_path="<model_artifact_path>", flavor=mlflow.sklearn, training_set=training_set, registered_model_name="<catalog_name>.<schema_name>.<model_name>", infer_input_example=True, # 自动根据训练数据生成 input_example 与模型签名 )

注意：关于 log_model 需遵守：

注意：1. 必须用 fe.log_model，不能用 mlflow.sklearn.log_model：只有前者会把 FeatureSpec 打包进 MLmodel，后者会导致 score_batch 找不到特征来源。

注意：2. 必须传 training_set：模型签名与 input_example 由 SDK 根据 training_set.feature_spec 自动推断，对齐的是特征表中的原始列。

注意：3. infer_input_example=True 自动从训练数据生成 input_example 与签名，是部署在线推理服务的前置条件，务必开启；不要再手动 infer_signature(…) 或手动传 input_example=…，否则会与 feature_spec 的原始列错位、污染契约。

注意：4. 所有预处理（OneHotEncoder、StandardScaler、自定义函数等）必须封装进 sklearn.Pipeline，传入的 model 是整条 Pipeline。不能在 Pipeline 外部用 pd.get_dummies / 手动 StandardScaler.fit_transform 后再 model.fit(…)——外部预处理步骤不会被序列化进 MLmodel，推理时 SDK 从特征表 lookup 出来的是原始 string 列，喂给只认 OHE 后列名的裸模型，轻则报 Feature names unseen at fit time，重则所有样本静默输出训练集均值。

注意：5. registered_model_name 建议使用三段式 catalog.schema.model_name，便于 Model Registry 治理。

fe.log_model 会在模型工件中写入一份 feature_spec.yaml，记录该模型在推理时所依赖的特征表、字段、主键与时间戳关系。这份文件将在后续的 score_batch 与在线推理中被自动读取。

十二、离线批量推理

完成训练后，调用 score_batch 进行离线批量推理。此时只需提供主键 DataFrame，特征列由 SDK 根据模型工件中的 feature_spec.yaml 自动从特征表中查找并补齐：

# 主键 DataFrame：仅包含主键 + 事件时间戳，不需要特征列# 注意：时间戳列需与特征表 schema 类型一致（如需 cast 为 timestamp）keys_df = spark.table("<catalog_name>.<schema_name>.<scoring_table_name>")result_df = fe.score_batch( model_uri="models:/<catalog_name>.<schema_name>.<model_name>/<version>", # 版本号对齐训练产出的最新版本 df=keys_df, result_type="double", # 可选，预测结果列的数据类型，默认 "double" timestamp_key="<timestamp_key>", # 训练含 timestamp_lookup_key 时强烈建议显式指定，保证 point-in-time 一致性)result_df.show()

注意：score_batch 的输入 df 有以下硬性约束：

注意：1. 不能包含 prediction 列，也不能是流式 DataFrame；

注意：2. 必须包含 feature_spec.yaml 中声明的所有 lookup_key；

注意：3. 训练时声明了 timestamp_lookup_key 的场景下，强烈建议在 df 中显式带上同名时间戳列以保证 point-in-time 一致性。若不带，SDK 会自动注入 current_timestamp() 取「当前最新可见」的特征，等同于「用今天的特征预测历史样本」，语义不严谨。

执行流程 ：

1. SDK 根据 model_uri 加载模型与 feature_spec.yaml。

2. 根据模型签名识别需要的特征列；对未在 keys_df 中提供的列，自动通过 Feature Lookup 从特征表中查找。

3. 拼装完整的输入 DataFrame 进入模型推理，输出预测结果。

十四、发布在线特征表

除页面操作外，也可在 Notebook 中通过 SDK 的 publish_table 一键发布：

fe.publish_table( catalog_name="<offline_catalog_name>", # 离线特征表所在 Catalog schema_name="<offline_schema_name>", # 离线特征表所在 Schema table_name="<offline_table_name>", # 离线特征表名 online_db_name="<online_schema_name>", # 在线表所在的库 online_table_name="<online_table_name>", # 在线表名 # trigger 不传则默认每天 0 点（Asia/Shanghai）全量同步一次)

注意：1. 调用 publish_table 要求工作空间下至少有：① 一个运行中的数据计算资源组；② 一个 PostgreSQL 类型连接。否则会抛 RuntimeError（错误信息中带 request_id 便于定位）。

注意：2. 如需更精细的调度（如每小时整点），可通过 trigger 参数自定义同步周期。

注意：补充说明（特定版本、环境） ：部分 SDK 版本或工作空间配置下，服务端会要求显式指定在线表所属的 Catalog（对应服务端 TargetPath.CatalogName）。若调用时报错 TargetPath.CatalogName 不能为空，可在上述参数基础上补传 online_catalog_name="<online_catalog_name>"（具体名称参考所在工作空间已配置的在线 Catalog）。

十五、删除在线特征表

当在线特征表不再需要时，可通过 SDK 删除：

fe.drop_online_table(name="<catalog_name>.<schema_name>.<table_name>")

注意：删除在线特征表会同时删除 PostgreSQL 中的在线表数据及对应的同步工作流，该操作不可恢复。删除前请确认没有在线推理服务正在依赖该在线表，否则服务会在下次请求时报特征查找失败。离线特征表不受影响。

十六、命名与组织规范

统一的命名与组织规范有助于团队协作、特征复用与权限治理。建议遵循以下约定：

此外建议：

·引用特征表统一使用三段式全限定名 catalog.schema.table，避免依赖默认 Catalog 带来的副作用，迁移到不同工作空间时也无需改代码。

·标签列建议用动词短语命名（如 order_purchase_delivery_time），避免与特征列重名。

十八、create_table 关键参数

十九、write_table 关键参数

20、get_table / read_table 参数

get_table 与 create_table 返回的均为 FeatureTable 实体（wedata.common.entities.feature_table.FeatureTable），由接口构造返回，不应由调用方直接实例化 。字段包括：name、table_id、description、primary_keys、partition_columns、features、timestamp_keys、creation_timestamp、online_stores、tags。

注意：tags 字段为惰性加载，未加载时直接访问会抛 ValueError。建议先确认表已正确返回后再读取标签。

21、set_feature_table_tag / delete_feature_table_tag 参数

22、FeatureLookup 关键参数

注意：FeatureLookup 共 9 个参数，其中包含 2 个已弃用参数（feature_name、output_name，请改用 feature_names、rename_outputs）。

23、create_training_set 关键参数

24、log_model 关键参数

注意：模型 signature 与 input_example 由 training_set.feature_spec 自动生成，无法通过参数覆盖 ，以保证 score_batch 时 lookup 列契约一致。请勿再手动 infer_signature(…) 或手动传 input_example=…。

25、score_batch 关键参数

26、publish_table 参数

注意：部分 SDK 版本、工作空间配置下，服务端会额外要求 online_catalog_name（在线表所属的 Catalog，对应服务端 TargetPath.CatalogName）。仅在报错 TargetPath.CatalogName 不能为空 时按需补传，取值参考所在工作空间已配置的在线 Catalog。

完整的调度参数详见 调度与触发器。

27、drop_table / drop_online_table 参数

28、相关文档

·界面操作

·最佳实践与异常排查

·模型列表

·部署在线推理服务

·全局数据源配置

·创建计算资源