El SDK de creación de modelos le permite desarrollar fórmulas de aprendizaje automático personalizadas y canalizaciones de funciones que se pueden utilizar en Adobe Experience Platform Data Science Workspace, que proporciona plantillas implementables en PySpark y Spark (Scala).
Este documento proporciona información sobre las distintas clases que se encuentran dentro del SDK de creación de modelos.
La clase DataLoader encapsula todo lo relacionado con la recuperación, el filtrado y la devolución de datos de entrada sin procesar. Algunos ejemplos de datos de entrada son los de formación, puntuación o ingeniería de funciones. Los cargadores de datos amplían la clase abstracta DataLoader
y debe anular 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 |
---|---|
Carga y devolución de datos de Platform como un DataFrame de Pandas |
|
Spark
En la tabla siguiente se describen los métodos abstractos de una Spark Clase del cargador de datos:
Método y descripción | Parámetros |
---|---|
Carga y devolución de datos de Platform como DataFrame |
|
El siguiente ejemplo recupera Platform 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
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
}
}
La clase DataSaver encapsula todo lo relacionado con el almacenamiento de datos de salida, incluidos los de la puntuación o la ingeniería de características. Los protectores de datos amplían la clase abstracta DataSaver
y debe anular el método abstracto save
.
PySpark
En la tabla siguiente se describen los métodos abstractos de una PySpark Clase del Data Saver:
Método y descripción | Parámetros |
---|---|
Recibir datos de salida como un DataFrame y almacenarlos en un conjunto de datos de Platform |
|
Spark (Scala)
En la tabla siguiente se describen los métodos abstractos de una Spark Clase del Data Saver:
Método y descripción | Parámetros |
---|---|
Recibir datos de salida como un DataFrame y almacenarlos en un conjunto de datos de Platform |
|
Para almacenar datos en un Platform conjunto de datos, las propiedades deben proporcionarse o definirse en el archivo de configuración:
Los siguientes ejemplos almacenan datos (prediction
) en un Platform conjunto de datos, donde el ID del conjunto de datos (datasetId
) e ID de 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()
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()
}
}
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 forma conjunta con una fábrica de canalizaciones de funciones para preparar los datos para la ingeniería de funciones.
PySpark
En la tabla siguiente se describen los métodos de clase de una clase de transformador de conjuntos de datos PySpark:
Método y descripción | Parámetros |
---|---|
compendio Toma un conjunto de datos como entrada y como salida un nuevo conjunto de datos derivado |
|
Spark (Scala)
En la tabla siguiente se describen los métodos abstractos de una Spark clase del transformador del conjunto de datos:
Método y descripción | Parámetros |
---|---|
Toma un conjunto de datos como entrada y como salida un nuevo conjunto de datos derivado |
|
La clase FeaturePipelineFactory contiene algoritmos de extracción de características y define las etapas de una canalización de características de principio a fin.
PySpark
En la tabla siguiente se describen los métodos de clase de una FeaturePipelineFactory de PySpark:
Método y descripción | Parámetros |
---|---|
compendio Crear y devolver una canalización de Spark que contenga una serie de transformadores de Spark |
|
compendio Recuperar y devolver el mapa del parámetro desde las propiedades de configuración |
|
Spark (Scala)
En la tabla siguiente se describen los métodos de clase de un Spark FeaturePipelineFactory:
Método y descripción | Parámetros |
---|---|
compendio Crear y devolver una canalización que contenga una serie de transformadores |
|
compendio Recuperar y devolver el mapa del parámetro desde las propiedades de configuración |
|
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 formación se definen en forma de Spark Canalización.
PySpark
En la tabla siguiente se describen los métodos de clase de una PipelineFactory de PySpark:
Método y descripción | Parámetros |
---|---|
compendio Crear y devolver una canalización de Spark que contenga la lógica y el algoritmo para la formación y puntuación de modelos. |
|
compendio Devuelva 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 |
|
compendio Puntúe utilizando el modelo entrenado y devuelva los resultados |
|
compendio Recuperar y devolver el mapa del parámetro desde las propiedades de configuración |
|
Spark (Scala)
En la tabla siguiente se describen los métodos de clase de un Spark PipelineFactory:
Método y descripción | Parámetros |
---|---|
compendio Crear y devolver una canalización que contenga la lógica y el algoritmo para la formación y la puntuación del modelo |
|
compendio Recuperar y devolver el mapa del parámetro desde las propiedades de configuración |
|
La clase MLEvaluator proporciona métodos para definir métricas de evaluación y determinar conjuntos de datos de prueba y aprendizaje.
PySpark
En la tabla siguiente se describen los métodos de clase de un MLEvaluator de PySpark:
Método y descripción | Parámetros |
---|---|
compendio Divide el conjunto de datos de entrada en subconjuntos de prueba y aprendizaje |
|
compendio Evalúa un modelo entrenado y devuelve los resultados de la evaluación |
|
Spark (Scala)
En la tabla siguiente se describen los métodos de clase de un Spark MLEvaluator:
Método y descripción | Parámetros |
---|---|
compendio Divide el conjunto de datos de entrada en subconjuntos de prueba y aprendizaje |
|
compendio Evalúa un modelo entrenado y devuelve los resultados de la evaluación |
|