Guide de développement du SDK

The Model Authoring SDK enables you to develop custom machine learning Recipes and Feature Pipelines which can be used in Adobe Experience Platform Data Science Workspace, providing implementable templates in PySpark and Spark (Scala).

Ce document fournit des informations sur les différentes classes trouvées dans le SDK de création de modèles.

DataLoader

La classe DataLoader englobe tous les éléments en lien avec la récupération, le filtrage et le renvoi de données d’entrée brutes. Les exemples de données d’entrée incluent les exemples de formation, de notation ou de conception des fonctionnalités. Les chargeurs de données étendent la classe abstraite DataLoader et doivent remplacer la méthode abstraite load.

PySpark

Le tableau suivant décrit les méthodes abstraites d’une classe Data Loader PySpark :

Méthode et description Paramètres

Méthode et description	Paramètres
`load(self, configProperties, spark)` Chargement et renvoi des données Platform sous la forme d’un cadre de données pandas	`self`: auto-référence `configProperties`: map des propriétés de configuration `spark`: session Spark

load(self, configProperties, spark)

Chargement et renvoi des données Platform sous la forme d’un cadre de données pandas

self: auto-référence
configProperties: map des propriétés de configuration
spark: session Spark

Spark

The following table describes the abstract methods of a Spark Data Loader class:

Méthode et description Paramètres

Méthode et description	Paramètres
`load(configProperties, sparkSession)` Chargement et renvoi des données Platform sous la forme d’un cadre de données	`configProperties`: map des propriétés de configuration `sparkSession`: session Spark

load(configProperties, sparkSession)

Chargement et renvoi des données Platform sous la forme d’un cadre de données

configProperties: map des propriétés de configuration
sparkSession: session Spark

Load data from a Platform dataset

The following example retrieves Platform data by ID and returns a DataFrame, where the dataset ID (datasetId) is a defined property in the configuration file.

PySpark

# PySpark

from sdk.data_loader import DataLoader

class MyDataLoader(DataLoader):
    """
    Implementation of DataLoader which loads a DataFrame and prepares data
    """

    def load_dataset(config_properties, spark, task_id):

        PLATFORM_SDK_PQS_PACKAGE = "com.adobe.platform.query"
        PLATFORM_SDK_PQS_INTERACTIVE = "interactive"

        # prepare variables
        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(task_id))

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

        # load dataset through Spark session

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

        # return as DataFrame
        return pd

Spark (Scala)

// Spark

package com.adobe.platform.ml

import java.time.LocalDateTime

import com.adobe.platform.ml.config.ConfigProperties
import com.adobe.platform.query.QSOption
import org.apache.spark.ml.feature.StringIndexer
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{StructType, TimestampType}
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.Column

/**
 * Implementation of DataLoader which loads a DataFrame and prepares data
 */
class MyDataLoader extends DataLoader {

    final val PLATFORM_SDK_PQS_PACKAGE: String = "com.adobe.platform.query"
    final val PLATFORM_SDK_PQS_INTERACTIVE: String = "interactive"
    final val PLATFORM_SDK_PQS_BATCH: String = "batch"

    /**
    *
    * @param configProperties - Configuration Properties map
    * @param sparkSession     - SparkSession
    * @return                 - DataFrame which is loaded for training
    */


  def load_dataset(configProperties: ConfigProperties, sparkSession: SparkSession, taskId: String): DataFrame = {

    require(configProperties != null)
    require(sparkSession != null)

    // Read the configs
    val serviceToken: String = sparkSession.sparkContext.getConf.get("ML_FRAMEWORK_IMS_ML_TOKEN", "").toString
    val userToken: String = sparkSession.sparkContext.getConf.get("ML_FRAMEWORK_IMS_TOKEN", "").toString
    val orgId: String = sparkSession.sparkContext.getConf.get("ML_FRAMEWORK_IMS_ORG_ID", "").toString
    val apiKey: String = sparkSession.sparkContext.getConf.get("ML_FRAMEWORK_IMS_CLIENT_ID", "").toString

    val dataSetId: String = configProperties.get(taskId).getOrElse("")

    // Load the dataset
    var df = sparkSession.read.format(PLATFORM_SDK_PQS_PACKAGE)
      .option(QSOption.userToken, userToken)
      .option(QSOption.serviceToken, serviceToken)
      .option(QSOption.imsOrg, orgId)
      .option(QSOption.apiKey, apiKey)
      .option(QSOption.mode, PLATFORM_SDK_PQS_INTERACTIVE)
      .option(QSOption.datasetId, dataSetId)
      .load()
    df.show()
    df
    }
}

DataSaver

La classe DataSaver englobe tous les éléments en lien avec le stockage des données de sortie, y compris les données d’évaluation ou de conception de fonctionnalités. Les Data Savers étendent la classe abstraite DataSaver et doivent remplacer la méthode abstraite save.

PySpark

The following table describes the abstract methods of a PySpark Data Saver class:

Méthode et description Paramètres

Méthode et description	Paramètres
`save(self, configProperties, dataframe)` Réception des données de sortie sous la forme d’un cadre de données et stockage dans un jeu de données Platform	`self`: auto-référence `configProperties`: map des propriétés de configuration `dataframe`: données à stocker sous la forme d’un cadre de données

save(self, configProperties, dataframe)

Réception des données de sortie sous la forme d’un cadre de données et stockage dans un jeu de données Platform

self: auto-référence
configProperties: map des propriétés de configuration
dataframe: données à stocker sous la forme d’un cadre de données

Spark (Scala)

The following table describes the abstract methods of a Spark Data Saver class:

Méthode et description Paramètres

Méthode et description	Paramètres
`save(configProperties, dataFrame)` Réception des données de sortie sous la forme d’un cadre de données et stockage dans un jeu de données Platform	`configProperties`: map des propriétés de configuration `dataFrame`: données à stocker sous la forme d’un cadre de données

save(configProperties, dataFrame)

Réception des données de sortie sous la forme d’un cadre de données et stockage dans un jeu de données Platform

configProperties: map des propriétés de configuration
dataFrame: données à stocker sous la forme d’un cadre de données

Save data to a Platform dataset

In order to store data onto a Platform dataset, the properties must be either provided or defined in the configuration file:

A valid Platform dataset ID to which data will be stored
Identifiant du client appartenant à votre organisation

Les exemples suivants stockent des données (predictionPlatform) dans un jeu de données , où l’identifiant de jeu de données (datasetId) et l’identifiant de client (tenantId) sont des propriétés définies dans le fichier de configuration.

PySpark

# PySpark

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


class MyDataSaver(DataSaver):
    """
    Implementation of DataSaver which stores a DataFrame to a Platform dataset
    """

    def save(self, config_properties, prediction):

        # Spark context
        sparkContext = prediction._sc

        # preliminary checks
        if config_properties is None:
            raise ValueError("config_properties parameter is null")
        if prediction is None:
            raise ValueError("prediction parameter is null")
        if sparkContext is None:
            raise ValueError("sparkContext parameter is null")
        
        PLATFORM_SDK_PQS_PACKAGE = "com.adobe.platform.query"

        # prepare variables
        scored_dataset_id = str(config_properties.get("scoringResultsDataSetId"))
        tenant_id = str(config_properties.get("tenant_id"))
        timestamp = "2019-01-01 00:00:00"

        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 ['service_token', 'user_token', 'org_id', 'scored_dataset_id', 'api_key', 'tenant_id']:
            if eval(arg) == 'None':
                raise ValueError("%s is empty" % arg)
        
        scored_df = prediction.withColumn("date", col("date").cast(StringType()))
        scored_df = scored_df.withColumn(tenant_id, struct(col("date"), col("store"), col("prediction")))
        scored_df = scored_df.withColumn("timestamp", lit(timestamp).cast(TimestampType()))
        scored_df = scored_df.withColumn("_id", lit("empty"))
        scored_df = scored_df.withColumn("eventType", lit("empty")

        # store data into dataset

        query_options = get_query_options(sparkContext)

        scored_df.select(tenant_id, "_id", "eventType", "timestamp").write.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.datasetId(), scored_dataset_id) \
            .save()

Spark (Scala)

// Spark

package com.adobe.platform.ml

import com.adobe.platform.ml.config.ConfigProperties
import com.adobe.platform.ml.impl.Constants
import com.adobe.platform.ml.sdk.DataSaver
import com.adobe.platform.query.QSOption
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.TimestampType

/**
 * Implementation of DataSaver which stores a DataFrame to a Platform dataset
 */

class ScoringDataSaver extends DataSaver {

  final val PLATFORM_SDK_PQS_PACKAGE: String = "com.adobe.platform.query"
  final val PLATFORM_SDK_PQS_BATCH: String = "batch"

  /**
    * Method that saves the scoring data into a dataframe
    * @param configProperties  - Configuration Properties map
    * @param dataFrame         - Dataframe with the scoring results
    */
    
  override def save(configProperties: ConfigProperties, dataFrame: DataFrame): Unit =  {

    require(configProperties != null)
    require(dataFrame != null)

    val predictionColumn = configProperties.get(Constants.PREDICTION_COL).getOrElse(Constants.DEFAULT_PREDICTION)
    val sparkSession = dataFrame.sparkSession

    val serviceToken: String = sparkSession.sparkContext.getConf.get("ML_FRAMEWORK_IMS_ML_TOKEN", "").toString
    val userToken: String = sparkSession.sparkContext.getConf.get("ML_FRAMEWORK_IMS_TOKEN", "").toString
    val orgId: String = sparkSession.sparkContext.getConf.get("ML_FRAMEWORK_IMS_ORG_ID", "").toString
    val apiKey: String = sparkSession.sparkContext.getConf.get("ML_FRAMEWORK_IMS_CLIENT_ID", "").toString
    val tenantId:String = configProperties.get("tenantId").getOrElse("")
    val timestamp:String = "2019-01-01 00:00:00"

    val scoringResultsDataSetId: String = configProperties.get("scoringResultsDataSetId").getOrElse("")
    import sparkSession.implicits._

    var df = dataFrame.withColumn("date", $"date".cast("String"))

    var scored_df  = df.withColumn(tenantId, struct(df("date"), df("store"), df(predictionColumn)))
    scored_df = scored_df.withColumn("timestamp", lit(timestamp).cast(TimestampType))
    scored_df = scored_df.withColumn("_id", lit("empty"))
    scored_df = scored_df.withColumn("eventType", lit("empty"))

    scored_df.select(tenantId, "_id", "eventType", "timestamp").write.format(PLATFORM_SDK_PQS_PACKAGE)
      .option(QSOption.userToken, userToken)
      .option(QSOption.serviceToken, serviceToken)
      .option(QSOption.imsOrg, orgId)
      .option(QSOption.apiKey, apiKey)
      .option(QSOption.datasetId, scoringResultsDataSetId)
      .save()
    }
}

DatasetTransformer

La classe DatasetTransformer modifie et transforme la structure d’un jeu de données. The Sensei Machine Learning Runtime does not require this component to be defined, and is implemented based on your requirements.

En ce qui concerne les pipelines de fonctionnalités, les transformateurs de jeux de données peuvent être utilisés en association avec une fabrique de pipelines de caractéristiques afin de préparer les données pour la conception des fonctionnalités.

PySpark

Le tableau suivant décrit les méthodes d’une classe DatasetTransformer PySpark :

Méthode et description Paramètres

Méthode et description	Paramètres
abstract `transform(self, configProperties, dataset)` Utilise un jeu de données en entrée et émet un nouveau jeu de données dérivé en sortie	`self`: auto-référence `configProperties`: map des propriétés de configuration `dataset`: jeu de données d’entrée pour la transformation

abstract
transform(self, configProperties, dataset)

Utilise un jeu de données en entrée et émet un nouveau jeu de données dérivé en sortie

self: auto-référence
configProperties: map des propriétés de configuration
dataset: jeu de données d’entrée pour la transformation

Spark (Scala)

The following table describes the abstract methods of a Spark dataset transformer class:

Méthode et description Paramètres

Méthode et description	Paramètres
`transform(configProperties, dataset)` Utilise un jeu de données en entrée et émet un nouveau jeu de données dérivé en sortie	`configProperties`: map des propriétés de configuration `dataset`: jeu de données d’entrée pour la transformation

transform(configProperties, dataset)

Utilise un jeu de données en entrée et émet un nouveau jeu de données dérivé en sortie

configProperties: map des propriétés de configuration
dataset: jeu de données d’entrée pour la transformation

FeaturePipelineFactory

La classe FeaturePipelineFactory contient des algorithmes d’extraction de fonctionnalités et définit les étapes d’un pipeline de fonctionnalités du début à la fin.

PySpark

Le tableau suivant décrit les méthodes d’une classe FeaturePipelineFactory PySpark :

Méthode et description Paramètres

Méthode et description	Paramètres
abstract `create_pipeline(self, configProperties)` Création et renvoi d’un pipeline Spark contenant une série de transformateurs Spark	`self`: auto-référence `configProperties`: map des propriétés de configuration
abstract `get_param_map(self, configProperties, sparkSession)` Récupération et renvoi de la map param des propriétés de configuration	`self`: auto-référence `configProperties`: propriétés de configuration `sparkSession`: session Spark

abstract
create_pipeline(self, configProperties)

Création et renvoi d’un pipeline Spark contenant une série de transformateurs Spark

self: auto-référence
configProperties: map des propriétés de configuration

abstract
get_param_map(self, configProperties, sparkSession)

Récupération et renvoi de la map param des propriétés de configuration

self: auto-référence
configProperties: propriétés de configuration
sparkSession: session Spark

Spark (Scala)

The following table describes the class methods of a Spark FeaturePipelineFactory:

Méthode et description Paramètres

Méthode et description	Paramètres
abstract `createPipeline(configProperties)` Création et renvoi d’un pipeline contenant une série de transformateurs	`configProperties`: map des propriétés de configuration
abstract `getParamMap(configProperties, sparkSession)` Récupération et renvoi de la map param des propriétés de configuration	`configProperties`: propriétés de configuration `sparkSession`: session Spark

abstract
createPipeline(configProperties)

Création et renvoi d’un pipeline contenant une série de transformateurs

configProperties: map des propriétés de configuration

abstract
getParamMap(configProperties, sparkSession)

Récupération et renvoi de la map param des propriétés de configuration

configProperties: propriétés de configuration
sparkSession: session Spark

PipelineFactory

The PipelineFactory class encapsulates methods and definitions for model training and scoring, where training logic and algorithms are defined in the form of a Spark Pipeline.

PySpark

Le tableau suivant décrit les méthodes d’une classe PipelineFactory PySpark :

Méthode et description	Paramètres
abstract `apply(self, configProperties)` Création et renvoi d’un pipeline Spark contenant la logique et l’algorithme de formation et de notation des modèles	`self`: auto-référence `configProperties`: propriétés de configuration
abstract `train(self, configProperties, dataframe)` Renvoie un pipeline personnalisé contenant la logique et l’algorithme de formation d’un modèle. Cette méthode n’est pas nécessaire si un pipeline Spark est utilisé	`self`: auto-référence `configProperties`: propriétés de configuration `dataframe`: jeu de données de fonctionnalités pour la saisie de la formation
abstract `score(self, configProperties, dataframe, model)` Note à l’aide du modèle formé et renvoie les résultats	`self`: auto-référence `configProperties`: propriétés de configuration `dataframe`: jeu de données d’entrée pour la notation `model`: modèle formé utilisé pour la notation
abstract `get_param_map(self, configProperties, sparkSession)` Récupération et renvoi de la map param des propriétés de configuration	`self`: auto-référence `configProperties`: propriétés de configuration `sparkSession`: session Spark

Spark (Scala)

The following table describes the class methods of a Spark PipelineFactory:

Méthode et description Paramètres

Méthode et description	Paramètres
abstract `apply(configProperties)` Création et renvoi d’un pipeline contenant la logique et l’algorithme de formation et de notation des modèles	`configProperties`: propriétés de configuration
abstract `getParamMap(configProperties, sparkSession)` Récupération et renvoi de la map param des propriétés de configuration	`configProperties`: propriétés de configuration `sparkSession`: session Spark

abstract
apply(configProperties)

Création et renvoi d’un pipeline contenant la logique et l’algorithme de formation et de notation des modèles

configProperties: propriétés de configuration

abstract
getParamMap(configProperties, sparkSession)

Récupération et renvoi de la map param des propriétés de configuration

configProperties: propriétés de configuration
sparkSession: session Spark

MLEvaluator

La classe MLEvaluator fournit des méthodes pour définir des mesures d’évaluation et déterminer des jeux de données de formation et de test.

PySpark

Le tableau suivant décrit les méthodes d’une classe MLEvaluator PySpark :

Méthode et description Paramètres

Méthode et description	Paramètres
abstract `split(self, configProperties, dataframe)` Divise le jeu de données d’entrée en sous-ensembles de formation et de test	`self`: auto-référence `configProperties`: propriétés de configuration `dataframe`: jeu de données d’entrée à fractionner
abstract `evaluate(self, dataframe, model, configProperties)` Évalue un modèle formé et renvoie les résultats de l’évaluation	`self`: auto-référence `dataframe`: un cadre de données constitué de données de formation et de test `model`: un modèle formé `configProperties`: propriétés de configuration

abstract
split(self, configProperties, dataframe)

Divise le jeu de données d’entrée en sous-ensembles de formation et de test

self: auto-référence
configProperties: propriétés de configuration
dataframe: jeu de données d’entrée à fractionner

abstract
evaluate(self, dataframe, model, configProperties)

Évalue un modèle formé et renvoie les résultats de l’évaluation

self: auto-référence
dataframe: un cadre de données constitué de données de formation et de test
model: un modèle formé
configProperties: propriétés de configuration

Spark (Scala)

The following table describes the class methods of a Spark MLEvaluator:

Méthode et description Paramètres

Méthode et description	Paramètres
abstract `split(configProperties, data)` Divise le jeu de données d’entrée en sous-ensembles de formation et de test	`configProperties`: propriétés de configuration `data`: jeu de données d’entrée à fractionner
abstract `evaluate(configProperties, model, data)` Évalue un modèle formé et renvoie les résultats de l’évaluation	`configProperties`: propriétés de configuration `model`: un modèle formé `data`: un cadre de données constitué de données de formation et de test

abstract
split(configProperties, data)

Divise le jeu de données d’entrée en sous-ensembles de formation et de test

configProperties: propriétés de configuration
data: jeu de données d’entrée à fractionner

abstract
evaluate(configProperties, model, data)

Évalue un modèle formé et renvoie les résultats de l’évaluation

configProperties: propriétés de configuration
model: un modèle formé
data: un cadre de données constitué de données de formation et de test

Ressources sur adobe.com

Customer Intelligence Multivariate Testing Contribution Analysis Advertising Analytics Identity Management Marketing Analytics

Guide de développement du SDK

DataLoader

Load data from a Platform dataset

DataSaver

Save data to a Platform dataset

DatasetTransformer

FeaturePipelineFactory

PipelineFactory

MLEvaluator

Ressources sur adobe.com

Sur cette page

Voir suivant :