SDK de creación de modelos

El SDK de creación de modelos le permite desarrollar fórmulas de aprendizaje automático personalizadas y tuberías de funciones que se pueden utilizar en Adobe Experience Platform Área de trabajo de ciencias de datos, proporcionando plantillas implementables en PySpark y Spark (Scala).

Este documento proporciona información sobre las distintas clases que se encuentran en el SDK de creación de modelos.

DataLoader

La clase DataLoader encapsula todo lo relacionado con la recuperación, filtrado y devolución de datos de entrada sin procesar. Algunos ejemplos de datos de entrada son los de capacitación, puntuación o ingeniería de funciones. Los cargadores de datos extienden la clase abstracta DataLoader y deben sobrescribir el método abstracto load.

PySpark

En la tabla siguiente se describen los métodos abstractos de una clase PySpark Data Loader:

Método y descripción	Parámetros
load(self, configProperties, spark) Carga y devolución de datos de plataforma como un DataFrame de Pandas	self:: Referencia automática configProperties:: Asignación de propiedades de configuración spark:: Sesión de encendido

Spark

En la tabla siguiente se describen los métodos abstractos de una clase Spark Data Loader:

Método y descripción	Parámetros
load(configProperties, sparkSession) Carga y devolución de datos de plataforma como marco de datos	configProperties:: Asignación de propiedades de configuración sparkSession:: Sesión de encendido

Cargar datos desde un Platform conjunto de datos

El ejemplo siguiente recupera Platform datos por ID y devuelve un DataFrame, donde el ID del conjunto de datos (datasetId) es una propiedad definida en el archivo de configuración.

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

Chispa (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 clase DataSaver encapsula todo lo relacionado con el almacenamiento de datos de salida, incluidos los datos de puntuación o ingeniería de características. Los servidores de datos extienden la clase abstracta DataSaver y deben sobrescribir el método abstracto save.

PySpark

En la tabla siguiente se describen los métodos abstractos de una clase PySpark Data Saver:

Método y descripción	Parámetros
save(self, configProperties, dataframe) Recibir datos de salida como marco de datos y almacenarlos en un conjunto de datos de plataforma	self:: Referencia automática configProperties:: Asignación de propiedades de configuración dataframe:: Datos que se almacenarán en forma de marco de datos

Chispa (Scala)

En la tabla siguiente se describen los métodos abstractos de una clase Spark Data Saver:

Método y descripción	Parámetros
save(configProperties, dataFrame) Recibir datos de salida como marco de datos y almacenarlos en un conjunto de datos de plataforma	configProperties:: Asignación de propiedades de configuración dataFrame:: Datos que se almacenarán en forma de marco de datos

Guardar datos en un Platform conjunto de datos

Para almacenar datos en un conjunto de datos Platform, las propiedades deben proporcionarse o definirse en el archivo de configuración:

• Una ID de conjunto de datos Platform válida en la que se almacenarán los datos
• El ID de inquilino que pertenece a su organización

Los siguientes ejemplos almacenan datos (prediction) en un conjunto de datos Platform, donde el ID del conjunto de datos (datasetId) y el ID del inquilino (tenantId) son propiedades definidas dentro del archivo de configuración.

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

Chispa (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 clase DatasetTransformer modifica y transforma la estructura de un conjunto de datos. El Sensei Machine Learning Runtime no requiere que se defina este componente y se implementa según sus necesidades.

Con respecto a una canalización de funciones, los transformadores de conjuntos de datos se pueden utilizar de manera conjunta con una fábrica de canalizaciones de funciones para preparar datos para la ingeniería de funciones.

PySpark

La siguiente tabla describe los métodos de clase de una clase de transformador de conjuntos de datos PySpark:

Método y descripción	Parámetros
abstract transform(self, configProperties, dataset) Toma un conjunto de datos como entrada y genera un nuevo conjunto de datos derivado	self:: Referencia automática configProperties:: Asignación de propiedades de configuración dataset:: El conjunto de datos de entrada para la transformación

Chispa (Scala)

La siguiente tabla describe los métodos abstractos de una clase de transformador de conjuntos de datos Spark:

Método y descripción	Parámetros
transform(configProperties, dataset) Toma un conjunto de datos como entrada y genera un nuevo conjunto de datos derivado	configProperties:: Asignación de propiedades de configuración dataset:: El conjunto de datos de entrada para la transformación

FeaturePipelineFactory

La clase FeaturePipelineFactory contiene algoritmos de extracción de funciones y define las etapas de una tubería de funciones de inicio a fin.

PySpark

En la tabla siguiente se describen los métodos de clase de PySpark FeaturePipelineFactory:

Método y descripción	Parámetros
abstract create_pipeline(self, configProperties) Crear y devolver una tubería de spark que contenga una serie de transformadores de chispa	self:: Referencia automática configProperties:: Asignación de propiedades de configuración
abstract get_param_map(self, configProperties, sparkSession) Recuperar y devolver mapa de parámetros de las propiedades de configuración	self:: Referencia automática configProperties:: Propiedades de configuración sparkSession:: Sesión de encendido

Chispa (Scala)

La siguiente tabla describe los métodos de clase de un Spark FeaturePipelineFactory:

Método y descripción	Parámetros
abstract createPipeline(configProperties) Crear y devolver una canalización que contenga una serie de transformadores	configProperties:: Asignación de propiedades de configuración
abstract getParamMap(configProperties, sparkSession) Recuperar y devolver mapa de parámetros de las propiedades de configuración	configProperties:: Propiedades de configuración sparkSession:: Sesión de encendido

PipelineFactory

La clase PipelineFactory encapsula métodos y definiciones para la formación y puntuación de modelos, donde la lógica y los algoritmos de capacitación se definen en forma de una Spark canalización.

PySpark

En la tabla siguiente se describen los métodos de clase de PySpark PipelineFactory:

Método y descripción	Parámetros
abstract apply(self, configProperties) Crear y devolver una tubería de spark que contiene la lógica y el algoritmo para la formación y la puntuación de modelos	self:: Referencia automática configProperties:: Propiedades de configuración
abstract train(self, configProperties, dataframe) Devolver una canalización personalizada que contenga la lógica y el algoritmo para entrenar un modelo. Este método no es necesario si se utiliza una canalización de Spark	self:: Referencia automática configProperties:: Propiedades de configuración dataframe:: Conjunto de datos de funciones para entradas de formación
abstract score(self, configProperties, dataframe, model) Puntuación mediante el modelo entrenado y devolución de los resultados	self:: Referencia automática configProperties:: Propiedades de configuración dataframe:: Conjunto de datos de entrada para la puntuación model:: Un modelo entrenado utilizado para la puntuación
abstract get_param_map(self, configProperties, sparkSession) Recuperar y devolver mapa de parámetros de las propiedades de configuración	self:: Referencia automática configProperties:: Propiedades de configuración sparkSession:: Sesión de encendido

Chispa (Scala)

La tabla siguiente describe los métodos de clase de un Spark PipelineFactory:

Método y descripción	Parámetros
abstract apply(configProperties) Crear y devolver una canalización que contenga la lógica y el algoritmo para la formación y la puntuación de modelos	configProperties:: Propiedades de configuración
abstract getParamMap(configProperties, sparkSession) Recuperar y devolver mapa de parámetros de las propiedades de configuración	configProperties:: Propiedades de configuración sparkSession:: Sesión de encendido

MLEvaluator

La clase MLEvaluator proporciona métodos para definir métricas de evaluación y determinar conjuntos de datos de prueba y formación.

PySpark

En la tabla siguiente se describen los métodos de clase de PySpark MLEvaluator:

Método y descripción	Parámetros
abstract split(self, configProperties, dataframe) Divide el conjunto de datos de entrada en subconjuntos de prueba y formación	self:: Referencia automática configProperties:: Propiedades de configuración dataframe:: Conjunto de datos de entrada que se va a dividir
abstract evaluate(self, dataframe, model, configProperties) Evalúa un modelo capacitado y devuelve los resultados de la evaluación	self:: Referencia automática dataframe:: Un marco de datos que consta de datos de prueba y formación model:: Un modelo entrenado configProperties:: Propiedades de configuración

Chispa (Scala)

La tabla siguiente describe los métodos de clase de un Spark MLEvaluator:

Método y descripción	Parámetros
abstract split(configProperties, data) Divide el conjunto de datos de entrada en subconjuntos de prueba y formación	configProperties:: Propiedades de configuración data:: Conjunto de datos de entrada que se va a dividir
abstract evaluate(configProperties, model, data) Evalúa un modelo capacitado y devuelve los resultados de la evaluación	configProperties:: Propiedades de configuración model:: Un modelo entrenado data:: Un marco de datos que consta de datos de prueba y formación