Konwersja RDD do LabelledPoint

Question

Jeśli mam RDD z około 500 kolumnami i 200 milionami wierszy, a RDD.columns.indexOf("target", 0) pokazuje Int = 77, która mówi mi, że moja docelowa zmienna zależna znajduje się w kolumnie numer 77. Ale nie mam wystarczającej wiedzy na temat wybierania pożądane (częściowe) kolumny jako cechy (np. chcę kolumny od 23 do 59, od 111 do 357, od 399 do 489). Zastanawiam się, czy mogę zastosować takie:

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

Wszelkie sugestie lub wskazówki będą mile widziane.

Może spartaczyłem RDD z DataFRrame, mogę przekonwertować rdd na DataFrame z .toDF() lub łatwiej jest osiągnąć cel z DataFrame niż RDD.

Answer 1

Zakładam dane wygląda mniej więcej tak:

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 

Więc mamy dane jak poniżej:

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

i chcemy ignorować foo i x2 i wyodrębnić 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) 
))