Spark中的平均單詞長度

Question

我有一個值列表以及它們作爲一個數組出現的所有事件的聚合長度。

例如：如果我的一句話就是

"I have a cat. The cat looks very cute" 

我的陣列看起來像

Array((I,1), (have,4), (a,1), (cat,6), (The, 3), (looks, 5), (very ,4), (cute,4)) 

現在我要計算每個單詞的平均長度。即發生的長度/次數。

我試圖做斯卡拉使用的編碼如下：

val avglen = arr.reduceByKey((x,y) => (x, y.toDouble/x.size.toDouble)) 

我得到一個錯誤，因爲在x.size如下 ^ 錯誤：值的大小是不是int

成員

請幫我在哪裏我錯了。

問候 VRK

Answer 1

您的評論後，我想我明白了：

val words = sc.parallelize(Array(("i", 1), ("have", 4), 
           ("a", 1), ("cat", 6), 
           ("the", 3), ("looks", 5), 
           ("very", 4), ("cute", 4))) 

val avgs = words.map { case (word, count) => (word, count/word.length.toDouble) } 

println("My averages are: ") 
avgs.take(100).foreach(println) 

假如你有這些詞的一段，你要計算的一段話的平均尺寸。

在兩個步驟中，用map-reduce方法和在spark-1.5.1：

val words = sc.parallelize(Array(("i", 1), ("have", 4), 
           ("a", 1), ("cat", 6), 
           ("the", 3), ("looks", 5), 
           ("very", 4), ("cute", 4))) 

val wordCount = words.map { case (word, count) => count}.reduce((a, b) => a + b) 
val wordLength = words.map { case (word, count) => word.length * count}.reduce((a, b) => a + b) 

println("The avg length is: " + wordLength/wordCount.toDouble) 

我使用連接到spark-kernel這一個.ipynb運行這個代碼是輸出。

Answer 2

如果我理解正確的問題：

val rdd: RDD[(String, Int) = ??? 
val ave: RDD[(String, Double) = 
    rdd.map { case (name, numOccurance) => 
     (name, name.length.toDouble/numOccurance) 
    } 

Answer 3

這是一個略顯混亂的問題。如果您的數據已經在Array[(String, Int)]集合中（可能在驅動程序的collect()之後），那麼您無需使用任何RDD轉換。事實上，還有你可以用fold*()跑過來集合搶平均一個漂亮的竅門：

val average = arr.foldLeft(0.0) { case (sum: Double, (_, count: Int)) => sum + count }/arr.foldLeft(0.0) { case (sum: Double, (word: String, count: Int)) => sum + count/word.length } 

的長篇大論類，但它本質上聚集在分子的總字符數，字數在數分母。在你的例子來看，我看到以下內容：

scala> val arr = Array(("I",1), ("have",4), ("a",1), ("cat",6), ("The", 3), ("looks", 5), ("very" ,4), ("cute",4)) 
arr: Array[(String, Int)] = Array((I,1), (have,4), (a,1), (cat,6), (The,3), (looks,5), (very,4), (cute,4)) 

scala> val average = ... 
average: Double = 3.111111111111111 

如果您有跨RDD[(String, Int)]分佈式您(String, Int)元組，你可以使用accumulators來很容易地解決這個問題：

val chars = sc.accumulator(0.0) 
val words = sc.accumulator(0.0) 
wordsRDD.foreach { case (word: String, count: Int) => 
    chars += count; words += count/word.length 
} 

val average = chars.value/words.value 

當在運行例如（放置在RDD）上面，我看到以下內容：

scala> val arr = Array(("I",1), ("have",4), ("a",1), ("cat",6), ("The", 3), ("looks", 5), ("very" ,4), ("cute",4)) 
arr: Array[(String, Int)] = Array((I,1), (have,4), (a,1), (cat,6), (The,3), (looks,5), (very,4), (cute,4)) 

scala> val wordsRDD = sc.parallelize(arr) 
wordsRDD: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[0] at parallelize at <console>:14 

scala> val chars = sc.accumulator(0.0) 
chars: org.apache.spark.Accumulator[Double] = 0.0 

scala> val words = sc.accumulator(0.0) 
words: org.apache.spark.Accumulator[Double] = 0.0 

scala> wordsRDD.foreach { case (word: String, count: Int) => 
    | chars += count; words += count/word.length 
    | } 
... 
scala>  val average = chars.value/words.value 
average: Double = 3.111111111111111