Criar um pipeline de machine learning do Apache Spark

  • Artigo

A biblioteca de machine learning dimensionável (MLlib) do Apache Spark traz capacidades de modelação para um ambiente distribuído. O pacote spark.ml do Spark é um conjunto de APIs de alto nível criadas em DataFrames. Estas APIs ajudam-no a criar e otimizar pipelines práticos de machine learning. Machine learning do Spark refere-se a esta API baseada em MLlib DataFrame e não à API de pipeline baseada em RDD mais antiga.

Um pipeline de machine learning (ML) é um fluxo de trabalho completo que combina múltiplos algoritmos de machine learning em conjunto. Podem ser necessários muitos passos para processar e aprender com os dados, exigindo uma sequência de algoritmos. Os pipelines definem as fases e a ordenação de um processo de machine learning. No MLlib, as fases de um pipeline são representadas por uma sequência específica de PipelineStages, em que um Transformador e um Avaliador executam tarefas.

Um Transformador é um algoritmo que transforma um DataFrame noutro através do transform() método . Por exemplo, um transformador de funcionalidades pode ler uma coluna de um DataFrame, mapeá-lo para outra coluna e produzir um novo DataFrame com a coluna mapeada anexada à mesma.

Um Avaliador é uma abstração de algoritmos de aprendizagem e é responsável por ajustar ou preparar um conjunto de dados para produzir um Transformador. Um Avaliador implementa um método com o nome fit(), que aceita um DataFrame e produz um DataFrame, que é um Transformador.

Cada instância sem estado de um Transformador ou de um Avaliador tem o seu próprio identificador exclusivo, que é utilizado ao especificar parâmetros. Ambos utilizam uma API uniforme para especificar estes parâmetros.

Exemplo de pipeline

Para demonstrar uma utilização prática de um pipeline de ML, este exemplo utiliza o ficheiro de dados de exemplo HVAC.csv que vem pré-carregado no armazenamento predefinido para o cluster do HDInsight, seja o Armazenamento do Azure ou Data Lake Storage. Para ver o conteúdo do ficheiro, navegue para o /HdiSamples/HdiSamples/SensorSampleData/hvac diretório. HVAC.csv contém um conjunto de horas com temperaturas de destino e reais para sistemas de AVAC (aquecimento, ventilação e ar condicionado) em vários edifícios. O objetivo é preparar o modelo nos dados e produzir uma temperatura prevista para um determinado edifício.

O seguinte código:

  1. Define um LabeledDocument, que armazena o BuildingID, SystemInfo (identificador e idade de um sistema) e um label (1,0 se o edifício estiver demasiado quente, 0,0 caso contrário).
  2. Cria uma função parseDocument de analisador personalizada que utiliza uma linha (linha) de dados e determina se o edifício está "quente" ao comparar a temperatura de destino com a temperatura real.
  3. Aplica o analisador ao extrair os dados de origem.
  4. Cria dados de preparação.
from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import HashingTF, Tokenizer
from pyspark.sql import Row

# The data structure (column meanings) of the data array:
# 0 Date
# 1 Time
# 2 TargetTemp
# 3 ActualTemp
# 4 System
# 5 SystemAge
# 6 BuildingID

LabeledDocument = Row("BuildingID", "SystemInfo", "label")

# Define a function that parses the raw CSV file and returns an object of type LabeledDocument


def parseDocument(line):
    values = [str(x) for x in line.split(',')]
    if (values[3] > values[2]):
        hot = 1.0
    else:
        hot = 0.0

    textValue = str(values[4]) + " " + str(values[5])

    return LabeledDocument((values[6]), textValue, hot)


# Load the raw HVAC.csv file, parse it using the function
data = sc.textFile(
    "wasbs:///HdiSamples/HdiSamples/SensorSampleData/hvac/HVAC.csv")

documents = data.filter(lambda s: "Date" not in s).map(parseDocument)
training = documents.toDF()

Este pipeline de exemplo tem três fases: Tokenizer e HashingTF (ambos Transformadores) e Logistic Regression (um Avaliador). Os dados extraídos e analisados no training DataFrame fluem através do pipeline quando pipeline.fit(training) são chamados.

  1. A primeira fase, Tokenizer, divide a SystemInfo coluna de entrada (que consiste no identificador do sistema e nos valores de idade) numa words coluna de saída. Esta nova words coluna é adicionada ao DataFrame.
  2. A segunda fase, HashingTF, converte a nova words coluna em vetores de funcionalidades. Esta nova features coluna é adicionada ao DataFrame. Estas duas primeiras fases são Transformadores.
  3. A terceira fase, LogisticRegression, é um Avaliador, pelo que o pipeline chama o LogisticRegression.fit() método para produzir um LogisticRegressionModel.
tokenizer = Tokenizer(inputCol="SystemInfo", outputCol="words")
hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol="features")
lr = LogisticRegression(maxIter=10, regParam=0.01)

# Build the pipeline with our tokenizer, hashingTF, and logistic regression stages
pipeline = Pipeline(stages=[tokenizer, hashingTF, lr])

model = pipeline.fit(training)

Para ver as novas words colunas e features adicionadas pelos Tokenizer transformadores e HashingTF e uma amostra do LogisticRegression avaliador, execute um PipelineModel.transform() método no DataFrame original. No código de produção, o próximo passo seria transmitir um DataFrame de teste para validar a preparação.

peek = model.transform(training)
peek.show()

# Outputs the following:
+----------+----------+-----+--------+--------------------+--------------------+--------------------+----------+
|BuildingID|SystemInfo|label|   words|            features|       rawPrediction|         probability|prediction|
+----------+----------+-----+--------+--------------------+--------------------+--------------------+----------+
|         4|     13 20|  0.0|[13, 20]|(262144,[250802,2...|[0.11943986671420...|[0.52982451901740...|       0.0|
|        17|      3 20|  0.0| [3, 20]|(262144,[89074,25...|[0.17511205617446...|[0.54366648775222...|       0.0|
|        18|     17 20|  1.0|[17, 20]|(262144,[64358,25...|[0.14620993833623...|[0.53648750722548...|       0.0|
|        15|      2 23|  0.0| [2, 23]|(262144,[31351,21...|[-0.0361327091023...|[0.49096780538523...|       1.0|
|         3|      16 9|  1.0| [16, 9]|(262144,[153779,1...|[-0.0853679939336...|[0.47867095324139...|       1.0|
|         4|     13 28|  0.0|[13, 28]|(262144,[69821,25...|[0.14630166986618...|[0.53651031790592...|       0.0|
|         2|     12 24|  0.0|[12, 24]|(262144,[187043,2...|[-0.0509556393066...|[0.48726384581522...|       1.0|
|        16|     20 26|  1.0|[20, 26]|(262144,[128319,2...|[0.33829638728900...|[0.58377663577684...|       0.0|
|         9|      16 9|  1.0| [16, 9]|(262144,[153779,1...|[-0.0853679939336...|[0.47867095324139...|       1.0|
|        12|       6 5|  0.0|  [6, 5]|(262144,[18659,89...|[0.07513008136562...|[0.51877369045183...|       0.0|
|        15|     10 17|  1.0|[10, 17]|(262144,[64358,25...|[-0.0291988646553...|[0.49270080242078...|       1.0|
|         7|      2 11|  0.0| [2, 11]|(262144,[212053,2...|[0.03678030020834...|[0.50919403860812...|       0.0|
|        15|      14 2|  1.0| [14, 2]|(262144,[109681,2...|[0.06216423725633...|[0.51553605651806...|       0.0|
|         6|       3 2|  0.0|  [3, 2]|(262144,[89074,21...|[0.00565582077537...|[0.50141395142468...|       0.0|
|        20|     19 22|  0.0|[19, 22]|(262144,[139093,2...|[-0.0769288695989...|[0.48077726176073...|       1.0|
|         8|     19 11|  0.0|[19, 11]|(262144,[139093,2...|[0.04988910033929...|[0.51246968885151...|       0.0|
|         6|      15 7|  0.0| [15, 7]|(262144,[77099,20...|[0.14854929135994...|[0.53706918109610...|       0.0|
|        13|      12 5|  0.0| [12, 5]|(262144,[89689,25...|[-0.0519932532562...|[0.48700461408785...|       1.0|
|         4|      8 22|  0.0| [8, 22]|(262144,[98962,21...|[-0.0120753606650...|[0.49698119651572...|       1.0|
|         7|      17 5|  0.0| [17, 5]|(262144,[64358,89...|[-0.0721054054871...|[0.48198145477106...|       1.0|
+----------+----------+-----+--------+--------------------+--------------------+--------------------+----------+

only showing top 20 rows

O model objeto pode agora ser utilizado para fazer predições. Para obter o exemplo completo desta aplicação de machine learning e instruções passo a passo para executá-la, veja Compilar aplicações de machine learning do Apache Spark no Azure HDInsight.

Ver também