지도

Question

I 아래와 같이 GroupByKey에서하던 RDD를 가질

(1 CompactBuffer (2.0, 3.0, 4.0)) (2 CompactBuffer (1.0, -1.0, -2.0))

그리고 I (1 * X_1^2 * 2 X_2^2, 3 * x_3^2)과 같아야

,

(1 CompactBuffer (4.0로 mapValues하고자 , 18.0, 48.0)), (2, CompactBuffer (1.0, 2.0, 12.0))

어떻게해야합니까?

도움 주셔서 감사합니다.

Answer 1

당신은 다음처럼 Stream(1, 2, ...)으로 zip를 사용하여 CompactBuffer 내용을 처리 할 수 ​​mapValues를 사용할 수 있습니다

val rdd = sc.parallelize(Seq(
    (1, 2.0), 
    (1, 3.0), 
    (1, 4.0), 
    (2, 1.0), 
    (2, -1.0), 
    (2, -2.0) 
)) 

val groupedRDD = rdd.groupByKey 
// res1: Array[(Int, Iterable[Double])] = Array(
// (1,CompactBuffer(2.0, 3.0, 4.0)), (2,CompactBuffer(1.0, -1.0, -2.0)) 
//) 

groupedRDD.mapValues(l => 
    l.zip(Stream from 1).map{ case (v, i) => v * v * i } 
) 
// res2: Array[(Int, Iterable[Double])] = Array(
// (1,List(4.0, 18.0, 48.0)), (2,List(1.0, 2.0, 12.0)) 
//)