스파크 SQL에서 RDD API로 변환

Question

스파크 SQL은 나에게 꽤 분명하다. 그러나, 나는 spark의 RDD API로 시작하고 있습니다. spark apply function to columns in parallel가 지적 하듯이 나를 의사 코드에서

def handleBias(df: DataFrame, colName: String, target: String = this.target) = { 
    val w1 = Window.partitionBy(colName) 
    val w2 = Window.partitionBy(colName, target) 

    df.withColumn("cnt_group", count("*").over(w2)) 
     .withColumn("pre2_" + colName, mean(target).over(w1)) 
     .withColumn("pre_" + colName, coalesce(min(col("cnt_group")/col("cnt_foo_eq_1")).over(w1), lit(0D))) 
     .drop("cnt_group") 
    } 
} 

에 대한 느린 셔플을 제거 할 수 있도록해야합니다 df foreach column (handleBias(column) 그래서 최소한의 데이터 프레임

val input = Seq(
    (0, "A", "B", "C", "D"), 
    (1, "A", "B", "C", "D"), 
    (0, "d", "a", "jkl", "d"), 
    (0, "d", "g", "C", "D"), 
    (1, "A", "d", "t", "k"), 
    (1, "d", "c", "C", "D"), 
    (1, "c", "B", "C", "D") 
) 
    val inputDf = input.toDF("TARGET", "col1", "col2", "col3TooMany", "col4") 

을로드하지만 제대로

를 매핑하는 데 실패

val rdd1_inputDf = inputDf.rdd.flatMap { x => {(0 until x.size).map(idx => (idx, x(idx)))}} 
     rdd1_inputDf.toDF.show 

이 질문에서 설명한 문제의 예제는 https://github.com/geoHeil/sparkContrastCodinghttps://github.com/geoHeil/sparkContrastCoding/blob/master/src/main/scala/ColumnParallel.scala입니다.

Answer 1

DataFrame에서 .rdd을 호출하면 RDD[Row]이 발생합니다.이 유형은 강력하게 입력되지 않습니다. 당신이 요소를 매핑 할 수 있도록하려면 당신은 을 통해 패턴 일치해야합니다

scala> val input = Seq(
    |  (0, "A", "B", "C", "D"), 
    |  (1, "A", "B", "C", "D"), 
    |  (0, "d", "a", "jkl", "d"), 
    |  (0, "d", "g", "C", "D"), 
    |  (1, "A", "d", "t", "k"), 
    |  (1, "d", "c", "C", "D"), 
    |  (1, "c", "B", "C", "D") 
    | ) 
input: Seq[(Int, String, String, String, String)] = List((0,A,B,C,D), (1,A,B,C,D), (0,d,a,jkl,d), (0,d,g,C,D), (1,A,d,t,k), (1,d,c,C,D), (1,c,B,C,D)) 

scala> val inputDf = input.toDF("TARGET", "col1", "col2", "col3TooMany", "col4") 
inputDf: org.apache.spark.sql.DataFrame = [TARGET: int, col1: string ... 3 more fields] 

scala> import org.apache.spark.sql.Row 
import org.apache.spark.sql.Row 

scala> val rowRDD = inputDf.rdd 
rowRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[3] at rdd at <console>:27 

scala> val typedRDD = rowRDD.map{case Row(a: Int, b: String, c: String, d: String, e: String) => (a,b,c,d,e)} 
typedRDD: org.apache.spark.rdd.RDD[(Int, String, String, String, String)] = MapPartitionsRDD[20] at map at <console>:29 

scala> typedRDD.keyBy(_._1).groupByKey.foreach{println} 
[Stage 7:>               (0 + 0)/4] 
(0,CompactBuffer((A,B,C,D), (d,a,jkl,d), (d,g,C,D))) 
(1,CompactBuffer((A,B,C,D), (A,d,t,k), (d,c,C,D), (c,B,C,D))) 

그렇지 않으면 당신이 사용할 수있는 Dataset 입력 : 나는 ㎖에이 사용하고자으로

scala> val ds = input.toDS 
ds: org.apache.spark.sql.Dataset[(Int, String, String, String, String)] = [_1: int, _2: string ... 3 more fields] 

scala> ds.rdd 
res2: org.apache.spark.rdd.RDD[(Int, String, String, String, String)] = MapPartitionsRDD[8] at rdd at <console>:30 

scala> ds.rdd.keyBy(_._1).groupByKey.foreach{println} 
[Stage 0:>               (0 + 0)/4] 
(0,CompactBuffer((0,A,B,C,D), (0,d,a,jkl,d), (0,d,g,C,D))) 
(1,CompactBuffer((1,A,B,C,D), (1,A,d,t,k), (1,d,c,C,D), (1,c,B,C,D)))