使用reduceByKey时scala中的类型不匹配

Question

我已经在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.\u 1作为键，这样我可以进一步使用join操作符，并且需要（键，值）对。但我的问题是，为什么在使用reduce函数时会有一个required类型？我认为这种形式是由我们自己设定的，而不是有规定的。我错了吗？

此外，如果我错了，那么为什么需要它（String，String，Int）？为什么不是别的？

ps：我知道（String，String，Int）是（a.\u 1 a.\u 2），（a.\u 1，a.\u 2，1））中的值类型，这是映射函数，但官方示例显示reduce函数（a，b）=




Answer 1

您的p7属于组织。阿帕奇。火花rdd。RDD[（String，（String，String，Int）]但是在你的reduceByKey中你使用了（a._1，（a._2，a._3b._3））类型（String，（String，Int））

p8的输出类型也应该是：org。阿帕奇。火花rdd。RDD[（String，（String，String，Int））]

所以像下面这样定义应该对你有用

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

您可以阅读我在pyspark中的回答，了解有关reduceByKey如何工作的更多详细信息

这个也应该有用

Answer 2

匿名用户

看看这些类型。按键减少是RDD[（K，V）]上的方法，带有签名：

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

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

在您的情况下，p7是

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

其中，K是字符串，V是（字符串，字符串，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。