使用“模型创作SDK”创建特征管道

重要

功能管道当前仅通过API可用。

Adobe Experience Platform允许您构建和创建自定义功能管道，通过Sensei机器学习框架运行时（以下简称“运行时”）大规模执行功能工程。

本文档介绍在功能管道中找到的各种类，并提供了使用PySpark中的模型创作SDK创建自定义功能管道的分步教程。

在运行特征管线时，将执行以下工作流：

菜谱将数据集加载到管道中。
对数据集进行功能转换并写入Adobe Experience Platform。
所转换的数据被加载用于培训。
特征流水线以梯度提升回归器为模型来定义阶段。
该管道用于拟合训练数据，并创建训练模型。
该模型与评分数据集进行转换。
然后，选择输出中有趣的列，并使用相关数据保存回Experience Platform。

入门指南

要在任何组织中运行菜谱，必须执行以下操作：

输入数据集。
数据集的模式。
转换的模式和基于该模式的空数据集。
输出模式和基于该模式的空数据集。

以上所有数据集都需要上传到Platform UI。要设置此设置，请使用Adobe提供的bootstrap脚本。

特征管线类

下表描述了构建特征管线时必须扩展的主要抽象类：

抽象类	描述
DataLoader	DataLoader类提供用于检索输入数据的实现。
DatasetTransformer	DatasetTransformer类提供转换输入数据集的实现。您可以选择不提供DatasetTransformer类，而是在FeaturePipelineFactory类中实现您的功能工程逻辑。
功能管道工厂	FeaturePipelineFactory类构建由一系列Spark Transporters组成的Spark Pipeline，以执行功能工程。您可以选择不提供FeaturePipelineFactory类，而是在DatasetTransformer类中实现您的功能工程逻辑。
数据保护程序	DataSaver类提供存储功能数据集的逻辑。

启动Feature Pipeline作业时，运行时首先执行DataLoader以将输入数据加载为DataFrame，然后通过执行DatasetTransformer、FeaturePipelineFactory或二者来修改DataFrame。最后，生成的特征数据集通过DataSaver进行存储。

以下流程图显示了运行时的执行顺序：

实现功能管线类

以下各节提供了有关为特征管道实现所需类的详细信息和示例。

在配置JSON文件中定义变量

配置JSON文件由键值对组成，旨在指定任何变量，以便在运行时稍后定义。这些键值对可以定义诸如输入数据集位置、输出数据集ID、租户ID、列标题等属性。

以下示例演示了在配置文件中找到的键值对：

配置JSON示例

[
    {
        "name": "fp",
        "parameters": [
            {
                "key": "dataset_id",
                "value": "000"
            },
            {
                "key": "featureDatasetId",
                "value": "111"
            },
            {
                "key": "tenantId",
                "value": "_tenantid"
            }
        ]
    }
]

您可以通过将config_properties定义为参数的任何类方法访问配置JSON。例如：

PySpark

dataset_id = str(config_properties.get(dataset_id))

有关更详细的配置示例，请参阅Data Science Workspace提供的pipeline.json文件。

使用DataLoader准备输入数据

DataLoader负责检索和过滤输入数据。 DataLoader的实现必须扩展抽象类DataLoader并覆盖抽象方法load。

下面的示例按ID检索Platform数据集，并将其作为DataFrame返回，其中数据集ID(dataset_id)是配置文件中定义的属性。

PySpark示例

# PySpark

from pyspark.sql.types import StringType, TimestampType
from pyspark.sql.functions import col, lit, struct
import logging

class MyDataLoader(DataLoader):
    def load_dataset(config_properties, spark, tenant_id, dataset_id):
    PLATFORM_SDK_PQS_PACKAGE = "com.adobe.platform.query"
    PLATFORM_SDK_PQS_INTERACTIVE = "interactive"

    service_token = str(spark.sparkContext.getConf().get("ML_FRAMEWORK_IMS_ML_TOKEN"))
    user_token = str(spark.sparkContext.getConf().get("ML_FRAMEWORK_IMS_TOKEN"))
    org_id = str(spark.sparkContext.getConf().get("ML_FRAMEWORK_IMS_ORG_ID"))
    api_key = str(spark.sparkContext.getConf().get("ML_FRAMEWORK_IMS_CLIENT_ID"))

    dataset_id = str(config_properties.get(dataset_id))

    for arg in ['service_token', 'user_token', 'org_id', 'dataset_id', 'api_key']:
        if eval(arg) == 'None':
            raise ValueError("%s is empty" % arg)

    query_options = get_query_options(spark.sparkContext)

    pd = spark.read.format(PLATFORM_SDK_PQS_PACKAGE) \
        .option(query_options.userToken(), user_token) \
        .option(query_options.serviceToken(), service_token) \
        .option(query_options.imsOrg(), org_id) \
        .option(query_options.apiKey(), api_key) \
        .option(query_options.mode(), PLATFORM_SDK_PQS_INTERACTIVE) \
        .option(query_options.datasetId(), dataset_id) \
        .load()
    pd.show()

    # Get the distinct values of the dataframe
    pd = pd.distinct()

    # Flatten the data
    if tenant_id in pd.columns:
        pd = pd.select(col(tenant_id + ".*"))

    return pd

使用DatasetTransformer转换数据集

DatasetTransformer提供用于转换输入DataFrame的逻辑并返回新的派生DataFrame。可以实现此类，以与FeaturePipelineFactory协同工作、作为唯一的特征工程组件工作，或者您可以选择不实现此类。

以下示例扩展了DatasetTransformer类：

PySpark示例

# PySpark

from sdk.dataset_transformer import DatasetTransformer
from pyspark.ml.feature import StringIndexer
from pyspark.sql.types import IntegerType
from pyspark.sql.functions import unix_timestamp, from_unixtime, to_date, lit, lag, udf, date_format, lower, col, split, explode
from pyspark.sql import Window
from .helper import setupLogger

class MyDatasetTransformer(DatasetTransformer):
    logger = setupLogger(__name__)

    def transform(self, config_properties, dataset):
        tenant_id = str(config_properties.get("tenantId"))

        # Flatten the data
        if tenant_id in dataset.columns:
            self.logger.info("Flatten the data before transformation")
            dataset = dataset.select(col(tenant_id + ".*"))
            dataset.show()

        # Convert isHoliday boolean value to Int
        # Rename the column to holiday and drop isHoliday
        pd = dataset.withColumn("holiday", col("isHoliday").cast(IntegerType())).drop("isHoliday")
        pd.show()

        # Get the week and year from date
        pd = pd.withColumn("week", date_format(to_date("date", "MM/dd/yy"), "w").cast(IntegerType()))
        pd = pd.withColumn("year", date_format(to_date("date", "MM/dd/yy"), "Y").cast(IntegerType()))

        # Convert the date to TimestampType
        pd = pd.withColumn("date", to_date(unix_timestamp(pd["date"], "MM/dd/yy").cast("timestamp")))

        # Convert categorical data
        indexer = StringIndexer(inputCol="storeType", outputCol="storeTypeIndex")
        pd = indexer.fit(pd).transform(pd)

        # Get the WeeklySalesAhead and WeeklySalesLag column values
        window = Window.orderBy("date").partitionBy("store")
        pd = pd.withColumn("weeklySalesLag", lag("weeklySales", 1).over(window)).na.drop(subset=["weeklySalesLag"])
        pd = pd.withColumn("weeklySalesAhead", lag("weeklySales", -1).over(window)).na.drop(subset=["weeklySalesAhead"])
        pd = pd.withColumn("weeklySalesScaled", lag("weeklySalesAhead", -1).over(window)).na.drop(subset=["weeklySalesScaled"])
        pd = pd.withColumn("weeklySalesDiff", (pd['weeklySales'] - pd['weeklySalesLag'])/pd['weeklySalesLag'])

        pd = pd.na.drop()
        self.logger.debug("Transformed dataset count is %s " % pd.count())

        # return transformed dataframe
        return pd

使用FeaturePipelineFactory工程数据功能

FeaturePipelineFactory允许您通过Spark Pipeline定义一系列Spark Transporters并将其链接到一起，从而实施您的功能工程逻辑。可以实现此类，以与DatasetTransformer协作，作为唯一的特征工程组件，或选择不实现此类。

下面的示例扩展了FeaturePipelineFactory类：

PySpark示例

# PySpark

from pyspark.ml import Pipeline
from pyspark.ml.regression import GBTRegressor
from pyspark.ml.feature import VectorAssembler

import numpy as np

from sdk.pipeline_factory import PipelineFactory

class MyFeaturePipelineFactory(FeaturePipelineFactory):

    def apply(self, config_properties):
        if config_properties is None:
            raise ValueError("config_properties parameter is null")

        tenant_id = str(config_properties.get("tenantId"))
        input_features = str(config_properties.get("ACP_DSW_INPUT_FEATURES"))

        if input_features is None:
            raise ValueError("input_features parameter is null")
        if input_features.startswith(tenant_id):
            input_features = input_features.replace(tenant_id + ".", "")

        learning_rate = float(config_properties.get("learning_rate"))
        n_estimators = int(config_properties.get("n_estimators"))
        max_depth = int(config_properties.get("max_depth"))

        feature_list = list(input_features.split(","))
        feature_list.remove("date")
        feature_list.remove("storeType")

        cols = np.array(feature_list)

        # Gradient-boosted tree estimator
        gbt = GBTRegressor(featuresCol='features', labelCol='weeklySalesAhead', predictionCol='prediction',
                       maxDepth=max_depth, maxBins=n_estimators, stepSize=learning_rate)

        # Assemble the fields to a vector
        assembler = VectorAssembler(inputCols=cols, outputCol="features")

        # Construct the pipeline
        pipeline = Pipeline(stages=[assembler, gbt])

        return pipeline

    def train(self, config_properties, dataframe):
        pass

    def score(self, config_properties, dataframe, model):
        pass

    def getParamMap(self, config_properties, sparkSession):
        return None

使用DataSaver存储您的功能数据集

DataSaver负责将您生成的功能数据集存储到存储位置。您对DataSaver的实现必须扩展抽象类DataSaver并覆盖抽象方法save。

以下示例扩展了按ID将数据存储到Platform数据集的DataSaver类，其中数据集ID(featureDatasetId)和租户ID(tenantId)是配置中定义的属性。

PySpark示例

# PySpark

from sdk.data_saver import DataSaver
from pyspark.sql.types import StringType, TimestampType
from pyspark.sql.functions import col, lit, struct


class MyDataSaver(DataSaver):
    def save(self, configProperties, data_feature):

        # Spark context
        sparkContext = data_features._sc

        # preliminary checks
        if configProperties is None:
            raise ValueError("configProperties parameter is null")
        if data_features is None:
            raise ValueError("data_features parameter is null")
        if sparkContext is None:
            raise ValueError("sparkContext parameter is null")

        # prepare variables
        timestamp = "2019-01-01 00:00:00"
        output_dataset_id = str(
            configProperties.get("featureDatasetId"))
        tenant_id = str(
            configProperties.get("tenantId"))
        service_token = str(
            sparkContext.getConf().get("ML_FRAMEWORK_IMS_ML_TOKEN"))
        user_token = str(
            sparkContext.getConf().get("ML_FRAMEWORK_IMS_TOKEN"))
        org_id = str(
            sparkContext.getConf().get("ML_FRAMEWORK_IMS_ORG_ID"))
        api_key = str(
            sparkContext.getConf().get("ML_FRAMEWORK_IMS_CLIENT_ID"))

        # validate variables
        for arg in ['output_dataset_id', 'tenant_id', 'service_token', 'user_token', 'org_id', 'api_key']:
            if eval(arg) == 'None':
                raise ValueError("%s is empty" % arg)

        # create and prepare DataFrame with valid columns
        output_df = data_features.withColumn("date", col("date").cast(StringType()))
        output_df = output_df.withColumn(tenant_id, struct(col("date"), col("store"), col("features")))
        output_df = output_df.withColumn("timestamp", lit(timestamp).cast(TimestampType()))
        output_df = output_df.withColumn("_id", lit("empty"))
        output_df = output_df.withColumn("eventType", lit("empty"))

        # store data into dataset
        output_df.select(tenant_id, "_id", "eventType", "timestamp") \
            .write.format("com.adobe.platform.dataset") \
            .option('orgId', org_id) \
            .option('serviceToken', service_token) \
            .option('userToken', user_token) \
            .option('serviceApiKey', api_key) \
            .save(output_dataset_id)

在应用程序文件中指定实现的类名

既然定义并实现了功能管线类，则必须在应用程序YAML文件中指定类的名称。

下面的示例指定实现的类名：

PySpark示例

#Name of the class which contains implementation to get the input data.
feature.dataLoader: InputDataLoaderForFeaturePipeline

#Name of the class which contains implementation to get the transformed data.
feature.dataset.transformer: MyDatasetTransformer

#Name of the class which contains implementation to save the transformed data.
feature.dataSaver: DatasetSaverForTransformedData

#Name of the class which contains implementation to get the training data
training.dataLoader: TrainingDataLoader

#Name of the class which contains pipeline. It should implement PipelineFactory.scala
pipeline.class: TrainPipeline

#Name of the class which contains implementation for evaluation metrics.
evaluator: Evaluator
evaluateModel: True

#Name of the class which contains implementation to get the scoring data.
scoring.dataLoader: ScoringDataLoader

#Name of the class which contains implementation to save the scoring data.
scoring.dataSaver: MyDatasetSaver

使用API创建您的功能管道引擎

既然您已经创作了功能管道，您需要创建一个Docker图像，以调用Sensei Machine Learning API中的功能管道端点。要调用功能管线端点，您需要Docker图像URL。

小贴士

如果您没有Docker URL，请访问将源文件打包到菜谱教程，以了解有关创建Docker主机URL的分步演练。

或者，您也可以使用以下Postman集合来帮助完成功能管道API工作流：

https://www.postman.com/collections/c5fc0d1d5805a5ddd41a

创建功能管线引擎

在您拥有Docker图像位置后，您可以通过对/engines执行POST，使用Sensei Machine Learning API创建功能管道引擎🔗。成功创建功能管道引擎可为您提供引擎唯一标识符(id)。请确保在继续之前保存此值。

创建MLInstance

使用新创建的engineID，您需要通过向/mlInstance端点发出POST请求，创建MLIstance。成功的响应返回一个有效负载，其中包含新创建的MLInstance的详细信息，包括在下一个API调用中使用的唯一标识符(id)。

创建实验

接下来，您需要创建Emperity。要创建实验，您需要具有MLIstance唯一标识符(id)并向/experiment端点发出POST请求。成功的响应返回一个有效负载，其中包含新创建的实验的详细信息，包括在下一个API调用中使用的唯一标识符(id)。

指定“实验”运行功能管线任务

创建实验后，必须将实验的模式更改为featurePipeline。要更改模式，请使用EXPERIMENT_ID对experiments/{EXPERIMENT_ID}/runs进行额外的POST，并在正文中发送{ "mode":"featurePipeline"}以指定功能管线“Emperice run”。

完成后，向/experiments/{EXPERIMENT_ID}发出GET请求以检索实验状态并等待实验状态更新以完成。

指定“实验”运行培训任务

接下来，您需要指定培训运行任务。将POST设置为experiments/{EXPERIMENT_ID}/runs，在正文中将模式设置为train并发送包含培训参数的任务数组。成功的响应返回包含所请求实验的详细信息的有效负荷。

完成后，向/experiments/{EXPERIMENT_ID}发出GET请求以检索实验状态并等待实验状态更新以完成。

指定“Emperice run scoring”任务

注意

要完成此步骤，您至少需要有一个成功的培训运行与您的实验关联。

成功运行培训后，您需要指定得分运行任务。将POST设置为experiments/{EXPERIMENT_ID}/runs，并在正文中将mode属性设置为“score”。这将开始您的得分实验运行。

完成后，向/experiments/{EXPERIMENT_ID}发出GET请求以检索实验状态并等待实验状态更新以完成。

评分完成后，您的功能管道应可运行。

后续步骤

通过阅读此文档，您使用“模型创作SDK”创作了一个功能管道，创建了一个Docker图像，并使用Docker图像URL通过使用Sensei Machine Learning API创建了功能管道模型。现在，您可以继续使用Sensei Machine Learning API大规模转换数据集和提取数据功能。

adobe.com 上的资源

Digital Foundation Ab Multivariate Testing Multichannel Communication First Party Data Mobile Marketing Tv Advertising

新增和改进的搜索

使用“模型创作SDK”创建特征管道

入门指南

特征管线类

实现功能管线类

在配置JSON文件中定义变量

使用DataLoader准备输入数据

使用DatasetTransformer转换数据集

使用FeaturePipelineFactory工程数据功能

使用DataSaver存储您的功能数据集

在应用程序文件中指定实现的类名

使用API创建您的功能管道引擎

创建功能管线引擎

创建MLInstance

创建实验

指定“实验”运行功能管线任务

指定“实验”运行培训任务

指定“Emperice run scoring”任务

后续步骤

adobe.com 上的资源

在此页面上

查看下一个：

学习

文档

社区

技术支持

资源

Adobe 帐户

Adobe