RDD to LabeledPoint Konvertierung

Question

Wenn ich eine RDD mit etwa 500 Spalten und 200 Millionen Zeilen habe, und RDD.columns.indexOf("target", 0) zeigt Int = 77, die mir sagt, dass meine gezielte abhängige Variable Spalte 77 ist. Aber ich habe nicht genug wissen, wie zu wählen gewünschte (Teil-) Spalten als Features (sagen wir Spalten von 23 bis 59, 111 bis 357, 399 bis 489). Ich frage mich, ob ich so etwas anwenden kann:

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

Alle Anregungen oder Anleitungen werden sehr geschätzt.

Vielleicht habe ich RDD mit DataFRRame vermasselt, kann ich die RDD in DataFrame mit .toDF() konvertieren oder es ist einfacher, das Ziel mit DataFrame als RDD zu erreichen.

Answer 1

Ich nehme an, Ihre Daten sieht mehr oder weniger wie folgt aus:

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 

So haben wir Daten wie folgt:

+----+------+----+----+----+ 
| 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| 
+----+------+----+----+----+ 

und wir wollen foo und x2 ignorieren und extrahieren 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) 
))