Pre-release
This release is a pre-release and may not be stable for production use.
Wedata Feature Engineering Library
Wedata 特征工程库，提供特征存储、特征管理、离在线同步等功能，适用于腾讯云 WeData 平台上的 ML 特征工程场景。
功能特性
- 特征表管理：创建、删除、修改特征表，支持 Iceberg 等多种存储引擎
- 特征数据读写：基于 Spark 的高效特征数据写入和读取
- 离在线同步：一键将离线特征表发布为在线特征表，支持周期性同步和一次性同步
- 在线特征服务：基于 Redis 的在线特征存储和实时查询
- 训练集构建：通过 Feature Lookup 自动关联多个特征表，构建训练数据集
安装
pip install tencent-wedata-feature-engineering
快速开始
from wedata.feature_store.feature_table_client import FeatureTableClient
# 初始化客户端（在 WeData Notebook 中自动获取环境配置）
client = FeatureTableClient(spark)
# 创建特征表
feature_table = client.create_table(
name="user_features",
database_name="my_feature_db",
primary_keys=["user_id"],
df=features_df,
timestamp_key="event_timestamp"
)
# 发布为在线特征表
client.publish_table(
table_name="user_features",
data_source_name="my_datasource"
)
快速开始（3.0 版本）
3.0 版本使用 FeatureEngineeringClient，需要 mlflow >= 3.0.0。
初始化客户端
from wedata.feature_engineering.client import FeatureEngineeringClient
# 初始化客户端（在 WeData Notebook 中自动获取 Spark 会话和环境配置）
fe = FeatureEngineeringClient()
# 也可以显式传入 SparkSession
# fe = FeatureEngineeringClient(spark)
创建数据库与特征表
from wedata.common.constants.engine_types import EngineTypes
# 创建特征数据库
fe.create_database("my_feature_db", catalog_name="DataLakeCatalog", comment="用户特征库")
# 创建特征表（通过 DataFrame 推断 schema）
feature_table = fe.create_table(
name="user_features",
database_name="my_feature_db",
catalog_name="DataLakeCatalog",
primary_keys=["user_id"],
timestamp_key="event_timestamp",
engine_type=EngineTypes.ICEBERG_ENGINE,
df=features_df,
description="用户画像特征表"
)
写入与读取特征数据
# 写入特征数据（默认追加模式）
fe.write_table(
name="user_features",
database_name="my_feature_db",
catalog_name="DataLakeCatalog",
df=new_features_df
)
# 读取特征表数据
df = fe.read_table(
name="user_features",
database_name="my_feature_db",
catalog_name="DataLakeCatalog"
)
df.show()
构建训练集
from wedata.common.entities.feature_lookup import FeatureLookup
# 定义特征查找
feature_lookups = [
FeatureLookup(
table_name="user_features",
lookup_key="user_id",
feature_names=["age", "gender", "purchase_count"],
timestamp_lookup_key="event_timestamp"
),
FeatureLookup(
table_name="item_features",
lookup_key="item_id",
feature_names=["category", "price"]
),
]
# 创建训练集
training_set = fe.create_training_set(
df=label_df,
feature_lookups=feature_lookups,
label="is_purchased",
exclude_columns=["user_id", "item_id"],
database_name="my_feature_db",
catalog_name="DataLakeCatalog"
)
# 加载训练 DataFrame
training_df = training_set.load_df()
模型训练与记录
import mlflow
import mlflow.sklearn
from sklearn.ensemble import GradientBoostingClassifier
# 训练模型
model = GradientBoostingClassifier()
model.fit(training_df.toPandas().drop("is_purchased", axis=1),
training_df.toPandas()["is_purchased"])
# 记录模型（自动关联特征元数据）
with mlflow.start_run():
fe.log_model(
model=model,
artifact_path="model",
flavor=mlflow.sklearn,
training_set=training_set,
registered_model_name="purchase_prediction_model"
)
批量推理
# 使用已注册的模型进行批量推理
predictions = fe.score_batch(
model_uri="models:/purchase_prediction_model/1",
df=inference_df
)
predictions.show()
发布在线特征表
from wedata.common.cloud_sdk_client.models import SchedulerConfiguration
# 一次性发布
fe.publish_table(
catalog_name="DataLakeCatalog",
schema_name="my_feature_db",
table_name="user_features",
online_db_name="online_feature_db",
online_table_name="user_features_online"
)
# 周期性同步（可选）
fe.publish_table(
catalog_name="DataLakeCatalog",
schema_name="my_feature_db",
table_name="user_features",
online_db_name="online_feature_db",
online_table_name="user_features_online",
trigger=SchedulerConfiguration(cron_expression="0 0 * * *") # 每天同步
)
管理特征表
# 获取特征表元数据
table_meta = fe.get_table(name="user_features", database_name="my_feature_db", catalog_name="DataLakeCatalog")
# 设置标签
fe.set_feature_table_tag("user_features", "my_feature_db", key="team", value="ml-platform")
# 删除标签
fe.delete_feature_table_tag("user_features", "my_feature_db", key="team")
# 删除在线表
fe.drop_online_table(catalog_name="DataLakeCatalog", schema_name="my_feature_db", table_name="user_features")
# 删除特征表
fe.drop_table(name="user_features", database_name="my_feature_db", catalog_name="DataLakeCatalog")
# 删除数据库
fe.drop_database("my_feature_db", catalog_name="DataLakeCatalog")
使用环境
本库需要在腾讯云 WeData 平台的 Notebook 或 CodeStudio 环境中使用，依赖平台提供的 Spark 引擎和 Feast 特征服务。
许可证
Apache 2.0
13 maintainers
Release files for tencent-wedata-feature-engineering 1.0.7.dev3
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| tencent_wedata_feature_engineering-1.0.7.dev3.tar.gz | 151.4 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| tencent_wedata_feature_engineering-1.0.7.dev3-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 312.2 kB
Release files / tencent_wedata_feature_engineering-1.0.7.dev3.tar.gz
| Download URL | tencent_wedata_feature_engineering-1.0.7.dev3.tar.gz |
|---|---|
| Size | 151.4 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
ff567ec38c7a400e87467364457fb4051b07e83c9e48110cdbae49f91aea5253
|
|
BLAKE2b-256 checksum How to use checksums |
137c20dfdfee817ca7613ad0e5012b7ebb9818e8f084dabc4c3f4c256fe79128
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.12.7
|
Release files / tencent_wedata_feature_engineering-1.0.7.dev3-py3-none-any.whl
| Download URL | tencent_wedata_feature_engineering-1.0.7.dev3-py3-none-any.whl |
|---|---|
| Size | 160.8 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
1a7fad0727f51e242e91c338cea7578a73189eb135a4dc4f88bde8cb090922fd
|
|
BLAKE2b-256 checksum How to use checksums |
e3c1edf25d25238167e6fe699989a7a63c3ff598084609ddab24fee04f813dd5
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.12.7
|
