Metadata-Version: 2.4
Name: tencent_wedata_feature_engineering
Version: 1.0.6
Summary: Wedata Feature Engineering Library Development
Home-page: 
Author: leondli
Author-email: 
License: Apache 2.0
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: pandas>=1.0.0
Requires-Dist: feast[postgres,redis]==0.49.0
Requires-Dist: grpcio==1.74.0
Requires-Dist: tencentcloud-sdk-python
Requires-Dist: ipython
Requires-Dist: mlflow-tclake-plugin==2.1.6
Requires-Dist: mlflow==2.17.2
Dynamic: author
Dynamic: classifier
Dynamic: description
Dynamic: description-content-type
Dynamic: license
Dynamic: requires-dist
Dynamic: requires-python
Dynamic: summary

# Wedata Feature Engineering Library

Wedata 特征工程库，提供特征存储、特征管理、离在线同步等功能，适用于腾讯云 WeData 平台上的 ML 特征工程场景。

## 功能特性

- **特征表管理**：创建、删除、修改特征表，支持 Iceberg 等多种存储引擎
- **特征数据读写**：基于 Spark 的高效特征数据写入和读取
- **离在线同步**：一键将离线特征表发布为在线特征表，支持周期性同步和一次性同步
- **在线特征服务**：基于 Redis 的在线特征存储和实时查询
- **训练集构建**：通过 Feature Lookup 自动关联多个特征表，构建训练数据集

## 安装

```bash
pip install tencent-wedata-feature-engineering
```

## 快速开始

```python
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`。

### 初始化客户端

```python
from wedata.feature_engineering.client import FeatureEngineeringClient

# 初始化客户端（在 WeData Notebook 中自动获取 Spark 会话和环境配置）
fe = FeatureEngineeringClient()

# 也可以显式传入 SparkSession
# fe = FeatureEngineeringClient(spark)
```

### 创建数据库与特征表

```python
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="用户画像特征表"
)
```

### 写入与读取特征数据

```python
# 写入特征数据（默认追加模式）
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()
```

### 构建训练集

```python
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()
```

### 模型训练与记录

```python
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"
    )
```

### 批量推理

```python
# 使用已注册的模型进行批量推理
predictions = fe.score_batch(
    model_uri="models:/purchase_prediction_model/1",
    df=inference_df
)
predictions.show()
```

### 发布在线特征表

```python
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 * * *")  # 每天同步
)
```

### 管理特征表

```python
# 获取特征表元数据
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
