Преобразование световых данных в формат org.apache.spark.rdd.RDD [org.apache.spark.mllib.linalg.Vector]

Question

Я очень новичок в scala и spark 2.1. Я пытаюсь вычислить корреляцию между многими элементами в кадре данных, который выглядит следующим образом:

item_1 | item_2 | item_3 | item_4 
    1 |  1 |  4 |  3 
    2 |  0 |  2 |  0 
    0 |  2 |  0 |  1 

Вот что я пробовал:

val df = sqlContext.createDataFrame(
    Seq((1, 1, 4, 3), 
     (2, 0, 2, 0), 
     (0, 2, 0, 1) 
).toDF("item_1", "item_2", "item_3", "item_4") 


val items = df.select(array(df.columns.map(col(_)): _*)).rdd.map(_.getSeq[Double](0)) 

И calcualte корреляции между элементами:

val correlMatrix: Matrix = Statistics.corr(items, "pearson") 

С сообщением об ошибке followning:

<console>:89: error: type mismatch; 
found : org.apache.spark.rdd.RDD[Seq[Double]] 
required: org.apache.spark.rdd.RDD[org.apache.spark.mllib.linalg.Vector] 
     val correlMatrix: Matrix = Statistics.corr(items, "pearson") 

Я не знаю, как создать org.apache.spark.rdd.RDD[org.apache.spark.mllib.linalg.Vector] из фрейма данных.

Это может быть очень простая задача, но я с ней борюсь, и я рад за любой совет.

Answer 1

Вы можете, например, использовать VectorAssembler. Собирают векторов и конвертировать в RDD

import org.apache.spark.ml.feature.VectorAssembler 

val rows = new VectorAssembler().setInputCols(df.columns).setOutputCol("vs") 
    .transform(df) 
    .select("vs") 
    .rdd 

Vectors Экстракт из :

• Спарк 1.x:

  rows.map(_.getAs[org.apache.spark.mllib.linalg.Vector](0)) 
  

• Спарк 2.x:

  rows 
      .map(_.getAs[org.apache.spark.ml.linalg.Vector](0)) 
      .map(org.apache.spark.mllib.linalg.Vectors.fromML) 
  

Что касается вашего кода:

• Вы Integer столбцов не Double.
• Данные не являются array, поэтому вы не можете использовать _.getSeq[Double](0).

Answer 2

Если ваша цель - выполнить корреляции pearson, вам действительно не нужно использовать RDD и Vectors. Ниже приведен пример выполнения коррекций pearson непосредственно на столбцах DataFrame (рассматриваемые столбцы являются типами Doubles).

Код:

import org.apache.spark.sql.{SQLContext, Row, DataFrame} 
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType, DoubleType} 
import org.apache.spark.sql.functions._ 


val rb = spark.read.option("delimiter","|").option("header","false").option("inferSchema","true").format("csv").load("rb.csv").toDF("name","beerId","brewerId","abv","style","appearance","aroma","palate","taste","overall","time","reviewer").cache() 

rb.agg(
    corr("overall","taste"), 
    corr("overall","aroma"), 
    corr("overall","palate"), 
    corr("overall","appearance"), 
    corr("overall","abv") 
    ).show() 

В этом примере я импортирование dataframe (с настраиваемым ограничителем, без заголовка и предполагаемыми типов данных), а затем просто выполняю функцию AGG против dataframe который имеет множество корреляций внутри него.

Выход:

+--------------------+--------------------+---------------------+-------------------------+------------------+ 
|corr(overall, taste)|corr(overall, aroma)|corr(overall, palate)|corr(overall, appearance)|corr(overall, abv)| 
+--------------------+--------------------+---------------------+-------------------------+------------------+ 
| 0.8762432795943761| 0.789023067942876| 0.7008942639550395|  0.5663593891357243|0.3539158620897098| 
+--------------------+--------------------+---------------------+-------------------------+------------------+ 

Как видно из результатов, (в целом, вкус) колонки имеют высокую корреляцию, в то время как (в целом, ABV) не так много.

Вот ссылка на Scala Docs DataFrame page which has the Aggregation Correlation Function.