特征工程 SDK

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

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

from wedata.feature_engineering.client import FeatureEngineeringClient
from wedata.common.constants.engine_types import EngineTypes

# DataBuddy Notebook 内核已预置 spark 会话；本地调试时需自行创建并传入
fe = FeatureEngineeringClient(spark)

# 幂等创建特征数据库（已存在则跳过）；支持单段名或二段式 catalog.schema
fe.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>")


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,
)

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

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

# 获取元数据
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)

# 设置或更新一个标签
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>",
)

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

from wedata.feature_engineering.client import FeatureEngineeringClient
from wedata.common.entities.feature_lookup import FeatureLookup

fe = 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()

import datetime
from wedata.common.entities.feature_lookup import FeatureLookup
from wedata.common.entities.feature_function import FeatureFunction

feature_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>",
    ),
]

import mlflow
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import GradientBoostingClassifier

with 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 与模型签名
    )

# 主键 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()

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）全量同步一次
)

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