OneHotEncoder в Spark Dataframe in Pipeline

Question

Я пытался запустить пример в Spark и Scala с помощью adult dataset.

Использование scala 2.11.8 и spark 1.6.1.

Проблема (в настоящее время) заключается в количестве категориальных признаков в этом наборе данных, которые все должны быть закодированы числами перед Спарк алгоритм ML может выполнять свою работу ..

До сих пор у меня есть это:

import org.apache.spark.ml.Pipeline 
import org.apache.spark.ml.classification.LogisticRegression 
import org.apache.spark.ml.feature.OneHotEncoder 
import org.apache.spark.sql.SQLContext 
import org.apache.spark.{SparkConf, SparkContext} 

object Adult { 
    def main(args: Array[String]): Unit = { 
    val conf = new SparkConf().setAppName("Adult example").setMaster("local[*]") 
    val sparkContext = new SparkContext(conf) 
    val sqlContext = new SQLContext(sparkContext) 

    val data = sqlContext.read 
     .format("com.databricks.spark.csv") 
     .option("header", "true") // Use first line of all files as header 
     .option("inferSchema", "true") // Automatically infer data types 
     .load("src/main/resources/adult.data") 

    val categoricals = data.dtypes filter (_._2 == "StringType") 
    val encoders = categoricals map (cat => new OneHotEncoder().setInputCol(cat._1).setOutputCol(cat._1 + "_encoded")) 
    val features = data.dtypes filterNot (_._1 == "label") map (tuple => if(tuple._2 == "StringType") tuple._1 + "_encoded" else tuple._1) 

    val lr = new LogisticRegression() 
     .setMaxIter(10) 
     .setRegParam(0.01) 
    val pipeline = new Pipeline() 
     .setStages(encoders ++ Array(lr)) 

    val model = pipeline.fit(training) 
    } 
} 

Однако это не работает. Вызов pipeline.fit по-прежнему содержит исходные функции строки и, таким образом, генерирует исключение. Как удалить эти столбцы "StringType" в конвейер? Или, может быть, я делаю это совершенно неправильно, поэтому, если у кого-то другое предложение, я доволен всем вводом :).

Причина, по которой я решил следовать этому потоку, состоит в том, что у меня обширный фон в Python и Pandas, но я пытаюсь изучить Scala и Spark.

Answer 1

Есть одна вещь, которая может быть довольно запутанной здесь, если вы привыкли к структурам более высокого уровня. Вы должны индексировать функции, прежде чем сможете использовать кодировщик. Как поясняется в the API docs:

один-горячий кодер (...) отображает столбец категории индексов в столбце двоичных векторов, с не более чем один, один-значение для каждой строки, что указывает на то, индекс категории ввода.

import org.apache.spark.ml.Pipeline 
import org.apache.spark.ml.feature.{StringIndexer, OneHotEncoder} 

val df = Seq((1L, "foo"), (2L, "bar")).toDF("id", "x") 

val categoricals = df.dtypes.filter (_._2 == "StringType") map (_._1) 

val indexers = categoricals.map (
    c => new StringIndexer().setInputCol(c).setOutputCol(s"${c}_idx") 
) 

val encoders = categoricals.map (
    c => new OneHotEncoder().setInputCol(s"${c}_idx").setOutputCol(s"${c}_enc") 
) 

val pipeline = new Pipeline().setStages(indexers ++ encoders) 

val transformed = pipeline.fit(df).transform(df) 
transformed.show 

// +---+---+-----+-------------+ 
// | id| x|x_idx|  x_enc| 
// +---+---+-----+-------------+ 
// | 1|foo| 1.0| (1,[],[])| 
// | 2|bar| 0.0|(1,[0],[1.0])| 
// +---+---+-----+-------------+ 

Как вы можете видеть, нет необходимости отказаться от строки столбцов из трубопровода. На практике OneHotEncoder будет принимать числовой столбец с NominalAttribute, BinaryAttribute или отсутствующим атрибутом типа.