스파크

Question

스파크 DataFrame을 응집 해제 (deaggregation) DF 스키마와 함께 제공됩니다 :

나는 "deaggdegated"열을 생산하기 위해 필요

id, agg_values 
432, 11 3.14 45 4.322 
984, 1 9.22 45 22.17 

:

id, item_id, value 
432, 11, 3.14 
432, 45, 4.322 
984, 1, 98.22 
984, 45, 222.1 

구문 분석 기능 :

을

private def parse_agg_scores(line: String): List[(String, String)] = { 
    val items: Array[String] = line.split(' ') 
    val ids = Iterator.from(0, 2).takeWhile(_ < items.size).map(items(_)) 
    val scores = Iterator.from(1, 2).takeWhile(_ < items.size).map(items(_)) 
    ids.zip(scores).toList 
    } 

flatMap을 시도했지만 작동하지 않습니다.

val res = df.flatMap{ row => 
     val line = row.getString(1) 

     parse_agg_scores(line) 
} 

Answer 1

그것은이 내가 생각해 솔루션입니다 sqlContext.implicits._ 수입하지 관련된 컴파일시 에러이었다에 말도없이 "가 작동하지 않습니다"

def unpivote_scores(df: DataFrame, colName: String, sqlContext: SQLContext): DataFrame = { 
    import sqlContext.implicits._ 

    df.flatMap { row => 
     val video_id = row.getInt(0) 
     val features = row.getString(1) 

     val items: Array[String] = features.split(' ') 
     val ids_string = Iterator.from(0, 2).takeWhile(_ < items.size).map(items(_)) 
     val scores_string = Iterator.from(1, 2).takeWhile(_ < items.size).map(items(_)) 

     val ids = ids_string.map(_.toInt) 
     val scores = scores_string.map(_.toDouble) 

     ids.zip(scores).toList.map(t => (video_id, t._1, t._2)) 
    }.toDF("VideoId", "FeatureId", "Score") 
    } 

Answer 2

당신은이 요소의 배열로 분할 4 요소의 배열에 새 열을 만들고, 그 열에 explode 수 - 후 별도의 컬럼으로이 두 요소의 배열을 분할 :

val result = dataFrame.withColumn("tuples", explode(array(
    array($"agg_values"(0), $"agg_values"(1)), 
    array($"agg_values"(2), $"agg_values"(3)) 
))) 
    .select($"id", $"tuples"(0) as "item_id", $"tuples"(1) as "value") 

result.show() 
// +---+-------+-----+ 
// | id|item_id|value| 
// +---+-------+-----+ 
// |432| 11.0| 3.14| 
// |432| 45.0|4.322| 
// |984| 1.0| 9.22| 
// |984| 45.0|22.17| 
// +---+-------+-----+ 

각 레코드 agg_values에 "쌍"의 다른 번호가있을 수 있습니다 경우에

편집, 우리는 폭발하기 전에 쌍 오전 배열로 배열을 변환하는 UDF가 필요합니다

// UDF to turn an array of Doubles into an Array of 2-item Arrays 
val groupPairs = udf { 
    arr: mutable.WrappedArray[Double] => arr.grouped(2).toArray 
} 

val result = dataFrame 
    .withColumn("pair", explode(groupPairs($"agg_values"))) 
    .select($"id", $"pair"(0) as "item_id", $"pair"(1) as "value")