Conversione da RDD a LabeledPoint

Question

Se si dispone di un RDD con circa 500 colonne e 200 milioni di righe e RDD.columns.indexOf("target", 0) mostra Int = 77 che indica che la variabile dipendente di destinazione si trova alla colonna numero 77. Ma non ho abbastanza conoscenza su come selezionare colonne (parziali) desiderate come caratteristiche (per esempio voglio colonne da 23 a 59, da 111 a 357, da 399 a 489). Mi chiedo se posso applicare tale:

val data = rdd.map(col => new LabeledPoint(
    col(77).toDouble, Vectors.dense(??.map(x => x.toDouble).toArray)) 

Eventuali suggerimenti o linee guida sarà molto apprezzato.

Forse ho incasinato RDD con DataFRrame, posso convertire il rdd in DataFrame con .toDF() oppure è più facile raggiungere l'obiettivo con DataFrame rispetto a RDD.

Answer 1

Presumo i dati sembra più o meno così:

import scala.util.Random.{setSeed, nextDouble} 
setSeed(1) 

case class Record(
    foo: Double, target: Double, x1: Double, x2: Double, x3: Double) 

val rows = sc.parallelize(
    (1 to 10).map(_ => Record(
     nextDouble, nextDouble, nextDouble, nextDouble, nextDouble 
    )) 
) 
val df = sqlContext.createDataFrame(rows) 
df.registerTempTable("df") 

sqlContext.sql(""" 
    SELECT ROUND(foo, 2) foo, 
     ROUND(target, 2) target, 
     ROUND(x1, 2) x1, 
     ROUND(x2, 2) x2, 
     ROUND(x2, 2) x3 
    FROM df""").show 

Così abbiamo i dati, come di seguito:

+----+------+----+----+----+ 
| foo|target| x1| x2| x3| 
+----+------+----+----+----+ 
|0.73| 0.41|0.21|0.33|0.33| 
|0.01| 0.96|0.94|0.95|0.95| 
| 0.4| 0.35|0.29|0.51|0.51| 
|0.77| 0.66|0.16|0.38|0.38| 
|0.69| 0.81|0.01|0.52|0.52| 
|0.14| 0.48|0.54|0.58|0.58| 
|0.62| 0.18|0.01|0.16|0.16| 
|0.54| 0.97|0.25|0.39|0.39| 
|0.43| 0.23|0.89|0.04|0.04| 
|0.66| 0.12|0.65|0.98|0.98| 
+----+------+----+----+----+ 

e vogliamo ignorare foo e x2 ed estrarre LabeledPoint(target, Array(x1, x3)):

// Map feature names to indices 
val featInd = List("x1", "x3").map(df.columns.indexOf(_)) 

// Or if you want to exclude columns 
val ignored = List("foo", "target", "x2") 
val featInd = df.columns.diff(ignored).map(df.columns.indexOf(_)) 

// Get index of target 
val targetInd = df.columns.indexOf("target") 

df.rdd.map(r => LabeledPoint(
    r.getDouble(targetInd), // Get target value 
    // Map feature indices to values 
    Vectors.dense(featInd.map(r.getDouble(_)).toArray) 
))