使用reduceByKey时在scala中类型不匹配答案

【问题标题】：type mismatch in scala while using reduceByKey使用reduceByKey时在scala中类型不匹配
【发布时间】：2018-05-03 12:31:38
【问题描述】：

我已经在 scala shell 中单独测试了我的错误代码

scala> val p6 = sc.parallelize(List( ("a","b"),("b","c")))
p6: org.apache.spark.rdd.RDD[(String, String)] = ParallelCollectionRDD[10] at parallelize at <console>:24

scala> val p7 = p6.map(a => ((a._1+a._2), (a._1, a._2, 1)))
p7: org.apache.spark.rdd.RDD[(String, (String, String, Int))] = MapPartitionsRDD[11] at map at <console>:26

scala> val p8 = p7.reduceByKey( (a,b) => (a._1,(a._2, a._3+b._3)))
<console>:28: error: type mismatch;
 found   : (String, (String, Int))
 required: (String, String, Int)
       val p8 = p7.reduceByKey( (a,b) => (a._1,(a._2, a._3+b._3)))

我想使用a._1作为键，这样我就可以进一步使用join操作符，并且必须是（键，值）对。但我的问题是，为什么在我使用缩减功能时会有required 类型？我认为格式是我们自己设定的，而不是受监管的。我错了吗？

另外，如果我错了，那为什么需要(String, String, Int)？为什么不是别的东西？

ps：我知道(String, String, Int)是map函数(a._1+a._2), (a._1, a._2, 1))中的值类型，但是官方例子显示reduce函数(a, b) => (a._1 + b._1, a._2 + b._2)是有效的。而且我认为所有这些包括我上面的代码都应该是有效的

【问题讨论】：

标签： scala apache-spark

【解决方案1】：

查看类型。按键归约是RDD[(K, V)] 上带有签名的方法：

def reduceByKey(func: (V, V) ⇒ V): RDD[(K, V)]

换句话说，输入参数和返回参数必须是同一类型。

在你的情况下p7 是

RDD[(String, (String, String, Int))]

其中K是String，V是(String, String, Int)，所以与reduceByKey一起使用的函数必须是

((String, String, Int), (String, String, Int)) => (String, String, Int)

一个有效的函数是：

p7.reduceByKey( (a,b) => (a._1, a._2, a._3 + b._3))

这会给你

(bc,(b,c,1))
(ab,(a,b,1))

结果。

如果要更改byKey 方法中的类型，则必须使用aggregateByKey 或combineByKey。

【讨论】：

谢谢！现在我知道了，规定reduce函数的输入和返回应该是一样的

【解决方案2】：

您的p7 属于p7: org.apache.spark.rdd.RDD[(String, (String, String, Int))]，但在您的reduceByKey 中，您使用了(a._1,(a._2, a._3+b._3))，其类型为(String, (String, Int))

p8的输出类型也应该是p8: org.apache.spark.rdd.RDD[(String, (String, String, Int))]

所以像下面这样定义应该适合你

val p8 = p7.reduceByKey( (a,b) => (a._1, a._2, a._3+b._3))

您可以阅读my answer in pyspark 了解有关 reduceByKey 工作原理的更多详细信息

和this one should help too

【讨论】：