Show Menu
SUJETS×

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