spark 用RDD怎麼合併連續相同的key？

1樓：屈靈翊

這問題的關鍵點在於你需要上下文或者說序列的資訊才能進行正確的聚合，而原來的資料並沒有提供序列資訊，所以像幾何題作輔助線一樣，你需要把序列的資訊加進去。在這裡我們可以考慮使用rdd的下標來標註key的位置，之後根據key來劃分相應的下標資訊，得到形如RDD[(Key, List[Index]]的結構，之後把每個List[Index]內相鄰的index合併起來，得到形如RDD[Key, List[(Start, End)]的結構，再將得到的rdd展平就能得到結果。val

srcData

:RDD

[(String, Int)]=

...//your data

valindexRdd

=srcData

.zipWithIndex

valresultWithReplicateKey=indexRdd

.groupByKey

.map

(aggregate

).collect

()//

利用下標資訊來聚合

接下來的問題就是如何使用下標資訊來聚合了，也就是實現aggregate函式。aggregate的乙個實現思路是對下標先進行排序，之後再進行foldLeft就可以了。這樣整個的時間複雜度是n / m * (m * lgm)，其中n是源資料的大小而m是以上List[Index]的大小，顯然太慢了。

接下來就是考慮怎麼優化的問題。

在每個partition內部，所有的下標都是連續的，可以很方便的聚合而無需對下標排序。因此我們可以考慮先在partition內部進行聚合，之後再根據下標資訊把不同partition返回的資訊再合併一次。那這樣的話，實現就只需要在@飛龍將的答案上稍微改一點就可以了，即初始化rdd時zipWithIndex一下，myfunc中保留一下index的資訊，最後collect之後再做一次合併。

這樣的時間複雜度應該在O(n)。

2樓：

scala的語法確實晦澀一些，但就你這個需求而言，還是很容易的defmain

(args

:Array

[String]):

Unit=)

.(myfunc).

collect

.foreach

(println)}

defmyfunc[T

](iter

:Iterator[T

]):Iterator

[(String, Int)]=

else

}res

=res.:

+(k1,

v1)res.

iterator

}輸出結果

(a,2)

(b,1)

(c,3)

(a,1)

(b,2)

3樓：Henry Ke

//初始化資料，沒用

scala> val a=Array("a","a","a","b","b","c","c","c","c","d")

a: Array[String] = Array(a, a, a, b, b, c, c, c, c, d)

scala> val b=a.map(f=>(f,1))

b: Array[(String, Int)] = Array((a,1), (a,1), (a,1), (b,1), (b,1), (c,1), (c,1), (c,1), (c,1), (d,1))

//上面是生成類似你的資料

scala> val rdd = sc.parallelize(b)

rdd: org.apache.

spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[2] at parallelize at :

31//這裡開始是你的問題

scala> val rddreduce=rdd.reduceByKey((a,b)=>a+b)

rddreduce: org.apache.

spark.rdd.RDD[(String, Int)] = ShuffledRDD[3] at reduceByKey at :

33//輸出結果

scala> rddreduce.collect.foreach(println)

(a,3)

(b,2)

(c,4)

(d,1)

spark 用RDD怎麼合併連續相同的key？

怎麼用matlab畫乙個牟合方蓋呢？

請問用Les Paul演奏布魯斯或者爵士合適嗎？

彩禮錢應該怎麼用獨自負擔月子中心的費用合理嗎？

其他用戶還看了：

spark 用RDD怎麼合併連續相同的key？

怎麼用matlab畫乙個牟合方蓋呢？

請問用Les Paul演奏布魯斯或者爵士合適嗎？

彩禮錢應該怎麼用 獨自負擔月子中心的費用合理嗎？

其他用戶還看了：

彩禮錢應該怎麼用獨自負擔月子中心的費用合理嗎？