spark scala 类型与 groupbykey 中的 zipwithIndex 不匹配

Question

我正在尝试测试 groupByKey 以找到某个主题的第 n 个最高分

我的数据是这样的

scala> a
res176: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[263] at map at <console>:51

scala> a.take(10).foreach{println}
(data science,DN,US,28,98,SMITH,data science)
(maths,DN,US,28,92,SMITH,maths)
(chemistry,DN,US,28,94,SMITH,chemistry)
(physics,DN,US,28,88,SMITH,physics)
(data science,DN,UK,25,93,JOHN,data science)
(maths,DN,UK,25,91,JOHN,maths)
(chemistry,DN,UK,25,95,JOHN,chemistry)
(physics,DN,UK,25,90,JOHN,physics)
(data science,DN,CA,29,67,MARK,data science)
(maths,DN,CA,29,68,MARK,maths)

scala>

所以第一行 "data science" 因为字符串是键，"DN,US,28,98,SMITH,data science" 是值作为字符串

现在我想通过

找到第二高的使用组

scala> a.groupByKey().flatMap(rec=>{ val max = rec._2.toList.map(x=>x.split(',')(3).toFloat).distinct.sortBy(x=>(-x)).zipWithIndex.filter(x=>x._2==2).toMap.keys
     | rec._2.toList.filter{x=>x.split(',')(3).toFloat==max}
     | }).take(15).foreach{println}

scala>

我在这里一无所获

如果我运行这个硬编码我得到值

scala> a.groupByKey().flatMap(rec=>{ val max = "98"
     | rec._2.toList.sortBy(x=>(-x.split(',')(3).toFloat)).takeWhile(rec=> max.contains(rec.split(',')(3)))}).take(15).foreach{println}
DN,IND,26,98,XMAN,maths
DPS,US,28,98,XOMAN,chemistry
DN,US,28,98,SMITH,data science

这也给了我价值

scala> a.groupByKey().flatMap(rec=>{ rec._2.toList.map(x=>x.split(',')(3).toFloat).distinct.sortBy(x=>(-x)).zipWithIndex.filter(x=>x._2==2).map(_._1)}).take(15).foreach{println}
94.0
92.0
95.0
93.0

一些更复杂的代码给了我输出

scala> a.groupByKey().flatMap(rec=>{ val max = rec._2.toList.map(x=>x.split(',')(3).toFloat).distinct.sortBy(x=>(-x)).take(1)
     | rec._2.toList.sortBy(x=>(-x.split(',')(3).toFloat)).takeWhile(rec=> max.contains(rec.split(',')(3).toFloat))}).take(15).foreach{println}
DN,IND,26,98,XMAN,maths
DPS,UK,25,96,SOMK,physics
DPS,US,28,98,XOMAN,chemistry
DN,US,28,98,SMITH,data science

当我使用 zipwithindex 时，看起来有些数据类型不匹配。有人可以帮我吗

Answer 1

由于 .toMap.keys，类型不匹配。在结果中，val max 是 Iterable[Float] 类型，因为 method keys returns Iterable[A]。

其中一个解决方案是在 max 计算结束时添加 head：

  val max = rec._2.toList
    .map(x => x.split(',')(3).toFloat)
    .distinct
    .sortBy(x => (-x))
    .zipWithIndex
    .filter(x => x._2 == 2)
    .toMap
    .keys
    .head

基本上，head 将 return 类型为 Float 的值。那么这段代码至少应该比较相等的类型 x.split(',')(3).toFloat == max.

不过，调用 head 并不是安全的方法。它可能会引发异常，如果在您的情况下 filter 函数可以 return 空列表。然后会抛出这样的异常：

java.util.NoSuchElementException: next on empty iterator

一旦它适用于具体的数据样本，您可以考虑重构此代码以与 Set 一起使用。而不是 head 做 .keys.toSet 并像使用 max.contains(rec.split(',')(3))

对其他示例所做的那样进行比较

spark scala 类型与 groupbykey 中的 zipwithIndex 不匹配

spark scala type mismatch with zipwithIndex in groupbykey

scala

flatmap

apache-spark