Acesso a dados usando o Spark no Data Science Workspace

O documento a seguir contém exemplos de como acessar dados usando o Spark para uso no Data Science Workspace. Para obter informações sobre como acessar dados usando notebooks JupyterLab, visite a documentação Acesso a dados dos blocos de anotações JupyterLab.

Introdução

O uso de Spark requer otimizações de desempenho que precisam ser adicionadas ao SparkSession. Além disso, também é possível configurar configProperties para ler e gravar posteriormente nos conjuntos de dados.

import com.adobe.platform.ml.config.ConfigProperties
import com.adobe.platform.query.QSOption
import org.apache.spark.sql.{DataFrame, SparkSession}

Class Helper {

 /**
   *
   * @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 = {
            // Read the configs
            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 sandboxName: String = sparkSession.sparkContext.getConf.get("sandboxName", "").toString

   }
}

Leitura de um conjunto de dados

Ao usar o Spark, você tem acesso a dois modos de leitura: interativo e em lote.

O modo interativo cria uma conexão JDBC (Java Database Connectivity) para Query Service e obtém resultados por meio de um JDBC regular ResultSet que é traduzido automaticamente para um DataFrame. Esse modo funciona de forma semelhante ao método Spark incorporado spark.read.jdbc(). Esse modo destina-se somente a conjuntos de dados pequenos. Se seu conjunto de dados exceder 5 milhões de linhas, sugerimos que você troque para o modo de lote.

O modo de lote usa o comando COPY de Query Service para gerar conjuntos de resultados de Parquet em um local compartilhado. Esses arquivos Parquet podem ser processados posteriormente.

Um exemplo da leitura de um conjunto de dados no modo interativo pode ser visto abaixo:

  // Read the configs
    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 sandboxName: String = sparkSession.sparkContext.getConf.get("sandboxName", "").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.imsOrg, orgId)
      .option(QSOption.apiKey, apiKey)
      .option(QSOption.mode, "interactive")
      .option(QSOption.datasetId, dataSetId)
      .option(QSOption.sandboxName, sandboxName)
      .load()
    df.show()
    df
  }

Da mesma forma, um exemplo de leitura de um conjunto de dados no modo de lote pode ser visto abaixo:

val df = sparkSession.read.format(PLATFORM_SDK_PQS_PACKAGE)
      .option(QSOption.userToken, userToken)
      .option(QSOption.imsOrg, orgId)
      .option(QSOption.apiKey, apiKey)
      .option(QSOption.mode, "batch")
      .option(QSOption.datasetId, dataSetId)
      .option(QSOption.sandboxName, sandboxName)
      .load()
    df.show()
    df

SELECIONAR colunas do conjunto de dados

df = df.select("column-a", "column-b").show()

Cláusula DISTINCT

A cláusula DISTINCT permite buscar todos os valores distintos em um nível de linha/coluna, removendo todos os valores duplicados da resposta.

Um exemplo de uso da função distinct() pode ser visto abaixo:

df = df.select("column-a", "column-b").distinct().show()

cláusula WHERE

O SDK Spark permite dois métodos de filtragem: Uso de uma expressão SQL ou filtragem por meio de condições.

Um exemplo do uso dessas funções de filtragem pode ser visto abaixo:

Expressão SQL

df.where("age > 15")

Condições do filtro

df.where("age" > 15 || "name" = "Steve")

Cláusula ORDER BY

A cláusula ORDER BY permite que os resultados recebidos sejam classificados por uma coluna especificada em uma ordem específica (crescente ou decrescente). No SDK Spark, isso é feito usando a função sort().

Um exemplo de uso da função sort() pode ser visto abaixo:

df = df.sort($"column1", $"column2".desc)

Cláusula LIMIT

A cláusula LIMIT permite limitar o número de registros recebidos do conjunto de dados.

Um exemplo de uso da função limit() pode ser visto abaixo:

df = df.limit(100)

Gravação em um conjunto de dados

Usando seu mapeamento configProperties, você pode gravar em um conjunto de dados no Experience Platform usando QSOption.

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 sandboxName: String = sparkSession.sparkContext.getConf.get("sandboxName", "").toString 

    df.write.format(PLATFORM_SDK_PQS_PACKAGE)
      .option(QSOption.userToken, userToken)
      .option(QSOption.imsOrg, orgId)
      .option(QSOption.apiKey, apiKey)
      .option(QSOption.datasetId, scoringResultsDataSetId)
      .option(QSOption.sandboxName, sandboxName)
      .save()

Próximas etapas

O Adobe Experience Platform Data Science Workspace fornece uma amostra de receita Scala (Spark) que usa as amostras de código acima para ler e gravar dados. Se quiser saber mais sobre como usar o Spark para acessar seus dados, consulte o Repositório GitHub Scala do Data Science.

Nesta página