tencent cloud

%pip install tencent-wedata-feature-engineering

from wedata.feature_store.client import FeatureStoreClient
# Build a feature engineering client instance

client = FeatureStoreClient(spark, cloud_secret_id=cloud_secret_id, cloud_secret_key=cloud_secret_key)

from wedata.feature_store.constants.engine_types import EngineTypes

# create user's feature table--EMR on hive
eature_table = client.create_table(
    name=table_name,  # Table name
    primary_keys=["wine_id"],      # Primary key
    df=wine_features_df,           # data
    engine_type=EngineTypes.HIVE_ENGINE,
    data_source_name=data_source_name,
    timestamp_key="event_timestamp",
    tags={ # Business tag
        "purpose": "demo", 
        "create_by": "wedata"
    }
)

# Create user feature table--DLC on iceberg (catalog not enabled)

feature_table = client.create_table(
    name=table_name,  # Table name
    database_name=database_name,
    primary_keys=["wine_id"],      # Primary key
    df=wine_features_df,           # data
    engine_type=EngineTypes.ICEBERG_ENGINE,
    data_source_name=data_source_name,
    timestamp_key="event_timestamp",
    tags={ # Business tag
        "purpose": "demo", 
        "create_by": "wedata"
    }
)

# Create user feature table--DLC on iceberg (catalog enabled)

feature_table = client.create_table(
    name=table_name,  # Table name
    primary_keys=["wine_id"],      # Primary key
    df=wine_features_df,           # data
    engine_type=EngineTypes.ICEBERG_ENGINE,
    data_source_name=data_source_name,
    timestamp_key="event_timestamp",
    tags={ # Business tag
        "purpose": "demo", 
        "create_by": "wedata"
    },
    catalog_name=catalog_name
)

from wedata.feature_store.constants.engine_types import EngineTypes
# Register feature table--EMR on hive

client.register_table(database_name=database_name, name=register_table_name, timestamp_key="event_timestamp",engine_type=EngineTypes.HIVE_ENGINE, data_source_name=data_source_name, primary_keys=["wine_id",])

# Register feature table--DLC on iceberg (catalog not enabled)

client.register_table(database_name=database_name, name=register_table_name, timestamp_key="event_timestamp",engine_type=EngineTypes.ICEBERG_ENGINE, data_source_name=data_source_name, primary_keys=["wine_id",])

# register feature table--DLC on iceberg(enable catalog)

client.register_table(database_name=database_name, name=register_table_name, timestamp_key="event_timestamp",engine_type=EngineTypes.HIVE_ENGINE, data_source_name=data_source_name, primary_keys=["wine_id",], catalog_name=catalog_name)

# Read offline feature table--EMR on hive--&--DLC on iceberg(catalog not enabled)

get_df = client.read_table(name=table_name, database_name=database_name)
get_df.show(30)

# Read online feature table--EMR on hive--&--DLC on iceberg(catalog not enabled)

online_config = RedisStoreConfig(host=redis_host, port=redis_password, db=redis_db, password=redis_password)
primary_keys_rows = [
    {"wine_id": 1},
    {"wine_id": 3}
]
result = client.read_table(name=table_name,database_name=database_name, is_online=True, online_config=online_config, entity_row=primary_keys_rows)
result.show()

# Read offline feature table--DLC on iceberg(enable catalog)

get_df = client.read_table(name=table_name, database_name=database_name, catalog_name=catalog_name)
get_df.show(30)

# Read online feature table--DLC on iceberg(enable catalog)

primary_keys_rows = [
    {"wine_id": 1},
    {"wine_id": 3}
]
result = client.read_table(name=table_name,database_name=database_name, is_online=True, online_config=online_config, entity_row=primary_keys_rows, catalog_name=catalog_name)
result.show()

# Get feature table data--EMR on hive--&--DLC on iceberg(catalog not enabled)

featureTable = client.get_table(name=table_name, database_name=database_name)

# Get online feature table data--DLC on iceberg(enable catalog)

featureTable = client.get_table(name=table_name, database_name=database_name, catalog_name=catalog_name)

# drop feature table data--EMR on hive--&--DLC on iceberg(disable catalog)

client.drop_table(table_name)


# drop feature table data--DLC on iceberg(enable catalog)
client.drop_table(table_name,catalog_name="DataLakeCatalog")

from wedata.feature_store.constants.constants import APPEND


# batch write--EMR on hive--&--DLC on iceberg(disable catalog)
client.write_table(
    name="user_features",
    df=user_data_df,
    database_name="feature_db",
    mode="append"
)


# strem write--EMR on hive--&--DLC on iceberg(disable catalog)
streaming_query = client.write_table(
    name="user_features",
    df=streaming_df,
    database_name="feature_db",
    checkpoint_location="/checkpoints/user_features",
    trigger={"processingTime": "10 seconds"}
)


# batch write--EMR on hive--&--DLC on iceberg(enable catalog)
client.write_table(
    name="user_features",
    df=user_data_df,
    database_name="feature_db",
    mode="append",
    catalog_name=catalog_name
)


# strem write--EMR on hive--&--DLC on iceberg(enable catalog)
streaming_query = client.write_table(
    name="user_features",
    df=streaming_df,
    database_name="feature_db",
    checkpoint_location="/checkpoints/user_features",
    trigger={"processingTime": "10 seconds"},
    catalog_name=catalog_name
)

from wedata.feature_store.entities.feature_lookup import FeatureLookup


# feature lookup--EMR on hive--&--DLC on iceberg(disable catalog)--&--DLC on iceberg(enable catalog)
wine_feature_lookup = FeatureLookup(
    table_name=table_name,
    lookup_key="wine_id",
    timestamp_lookup_key="event_timestamp"
)

from wedata.feature_store.entities.feature_lookup import FeatureLookup
from wedata.feature_store.entities.feature_function import FeatureFunction

# define feature lookup
wine_feature_lookup = FeatureLookup(
    table_name=table_name,
    lookup_key="wine_id",
    timestamp_lookup_key="event_timestamp"
)


# prepare training data
inference_data_df = wine_df.select(f"wine_id", "quality", "event_timestamp")


# create trainingset--EMR on hive--&--DLC on iceberg(disable catalog)
training_set = client.create_training_set(
    df=inference_data_df,  # basic data
    feature_lookups=[wine_feature_lookup],  # feature search configuration
    label="quality",  # tag column
    exclude_columns=["wine_id", "event_timestamp"]  # exclude unnecessary columns
)


# create trainingset--DLC on iceberg(enable catalog)
training_set = client.create_training_set(
    df=inference_data_df,  # basic data
    feature_lookups=[wine_feature_lookup],  # feature search configuration
    label="quality",  # tag column
    exclude_columns=["wine_id", "event_timestamp"],  # exclude unnecessary columns
    database_name=database_name,
    catalog_name=catalog_name
)


# get final training DataFrame
training_df = training_set.load_df()


# print trainingset data
print(f"\\n=== Training set data ===")
training_df.show(10, True)

# Training model

from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
from sklearn.ensemble import RandomForestClassifier
import mlflow.sklearn
import pandas as pd
import os 
project_id=os.environ["WEDATA_PROJECT_ID"]
mlflow.set_experiment(experiment_name=expirement_name)


# set mlflow tracking_uri
mlflow.set_tracking_uri("http://30.22.36.75:5000")


# convert Spark DataFrame to Pandas DataFrame to train
train_pd = training_df.toPandas()


# delete timestamp
# train_pd.drop('event_timestamp', axis=1)


# prepare features and tags
X = train_pd.drop('quality', axis=1)
y = train_pd['quality']


# split traningset and testset
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)


# or convert datatime to timestamp(second)
for col in X_train.select_dtypes(include=['datetime', 'datetimetz']):
    X_train[col] = X_train[col].astype('int64') // 10**9   # Convert nanoseconds to seconds


# Verify that no missing values cause the dtype to be downgraded to object.
X_train = X_train.fillna(X_train.median(numeric_only=True))


# Initialize, train and log the model.--EMR on hive--&--DLC on iceberg(disable catalog)--&--DLC on iceberg(enable catalog)
model = RandomForestClassifier(n_estimators=100, max_depth=3, random_state=42)
model.fit(X_train, y_train)

with mlflow.start_run():
    client.log_model(
        model=model,
        artifact_path="wine_quality_prediction",  # model artifact path
        flavor=mlflow.sklearn,
        training_set=training_set,
        registered_model_name=model_name,  # model name (if catalog is enabled, must be catalog model name)
    )

# run score_batch--EMR on hive--&--DLC on iceberg(disable catalog)--&--DLC on iceberg(enable catalog)

result = client.score_batch(model_uri=f"models:/{model_name}/1", df=wine_df)
result.show()

from wedata.feature_store.common.store_config.redis import RedisStoreConfig
from wedata.feature_store.cloud_sdk_client.models import TaskSchedulerConfiguration

online_config = RedisStoreConfig(host=redis_host, port=redis_password, db=redis_db, password=redis_password)


# publish a offline feature table to online feature table--EMR on hive--&--DLC on iceberg(disable catalog)


# Case 1 Pass the offline table table_name from the specified offline data source to the default online data source in snapshot mode.
result = client.publish_table(table_name=table_name, data_source_name=data_source_name, database_name=database_name,
                              is_cycle=False, is_use_default_online=True)


# Case 2 Sync the offline table table_name from the specified offline data source to the designated non-default data source in snapshot mode. Note that the IP address here must be consistent with the login IP information of the entered data source.
result = client.publish_table(name=table_name, data_source_name=data_source_name, database_name=database_name,
                               is_cycle=False, is_use_default_online=False, online_config=online_config)


# Case 3 Sync the offline table table_name from the specified offline data source to the default online data source in cycle mode, with a run interval of every five minutes.
cycle_obj = TaskSchedulerConfiguration()
cycle_obj.CrontabExpression = "0 0/5 * * * ?"
result = client.publish_table(name=table_name, data_source_name=data_source_name, database_name=database_name,
                              is_cycle=True, cycle_obj=cycle_obj, is_use_default_online=True)
                           


# publish a offline feature table to online feature table--DLC on iceberg(enable catalog)


# Case 1 Pass the offline table table_name from the specified offline data source to the default online data source in snapshot mode.
result = client.publish_table(table_name=table_name, data_source_name=data_source_name, database_name=database_name,
                              is_cycle=False, is_use_default_online=True, catalog_name=catalog_name)


# Case 2 Sync the offline table table_name from the specified offline data source to the designated non-default data source in snapshot mode. Note that the IP address here must be consistent with the login IP information of the entered data source.
result = client.publish_table(table_name=table_name, data_source_name=data_source_name, database_name=database_name,
                              is_cycle=False, is_use_default_online=False, online_config=online_config, catalog_name=catalog_name)


# Case 3 Sync the offline table table_name from the specified offline data source to the default online data source in cycle mode, with a run interval of every five minutes.
cycle_obj = TaskSchedulerConfiguration()
cycle_obj.CrontabExpression = "0 0/5 * * * ?"
result = client.publish_table(table_name=table_name, data_source_name=data_source_name, database_name=database_name,
                              is_cycle=True, cycle_obj=cycle_obj, is_use_default_online=True, catalog_name=catalog_name)