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
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file tencent_wedata_feature_engineering-1.0.6.tar.gz.
File metadata
- Download URL: tencent_wedata_feature_engineering-1.0.6.tar.gz
- Upload date:
- Size: 143.7 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.14.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
2dde9115729b0749fa1293d84fa03454cafb1fc0fe1c8882e10b802dff617896
|
|
| MD5 |
43f94076f7879603fd8fda2edd254f68
|
|
| BLAKE2b-256 |
5f4c2623ccada31d98d70f4318d499792197f86c900038ca547432eac907b8f6
|
File details
Details for the file tencent_wedata_feature_engineering-1.0.6-py3-none-any.whl.
File metadata
- Download URL: tencent_wedata_feature_engineering-1.0.6-py3-none-any.whl
- Upload date:
- Size: 152.8 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.14.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
1a228ced8f756ca9fa01d558555577a1877c55112d11a2c2ec0864dcee6f7eb7
|
|
| MD5 |
622d37940278d872d1328dac64039b80
|
|
| BLAKE2b-256 |
d6a5fa8c8c7ad121fb4e6e8e7e798bd1111599b51ec4d7b8b793770e1dd07c08
|
