在scala中使用reduceByKey和case

Question

我有这种类型；

 column1 column2 int 
 ((a,b),1)
 ((a,c),1)
 ((k,a),1)

我需要两种类型的结果，首先是所有 column2 的 column1

(a,total)

其次column1参数不等于column2参数

(a,total)

对于这种类型，我如何使用 ReduceByKey ？

我的代码：

var data = sc.textFile("tttt.tsv")
var satir = data.map(line=> ((line.split("\t")(1).toString,line.split("\t")(2).toString),1))

---关于第二个场景

我的数据类型示例是

column1 column2 int
a b,1 
a c,1
a a,1
a d,1

我需要 reduceByKey，因为在第二种情况下 column1 不等于 column1

例如，我的示例数据结果 = (a,b,1) + (a,c,1)+(a,d,1) = 3

Answer 1

如果我正确理解您的问题，以下是获得您所问问题的一种方法：

val rdd = sc.parallelize(Seq(
  (("a", "b"), 1),
  (("a", "c"), 1),
  (("a", "d"), 1),
  (("a", "a"), 1),
  (("k", "k"), 1),
  (("k", "a"), 1),
  (("k", "b"), 1)
))

val rdd1 = rdd.map{ case ((x, y), c) => (x, c) }.
  reduceByKey(_ + _)

scala> rdd1.collect.foreach(println)
(a,4)
(k,3)

val rdd2 = rdd.filter{ case ((x, y), c) => x != y }.
  map{ case ((x, y), c) => (x, c) }.
  reduceByKey(_ + _)

scala> rdd2.collect.foreach(println)
(a,3)
(k,2)

Answer 2

对于第一个场景，您可以使用这个。

val arrangedDF = satir.map(pairData => (pairData._1._1, (pairData._1._2, 1)))
val result1DF = arrangedDF.reduceByKey((x,y) => x._2 + y._2)

在这里我重新排列了数据的表示，我分离了键并将其余数据放入不同的元组中，您可以直接应用reduceByKey。

对于第二个场景，您可以使用它。

val result2DF = arrangedDF.filter( pairData => pairData._1 != pairData._2._1).reduceByKey((x,y) => x._2 + y._2)

在第二个场景中，我重用了arrangedDF，并针对您想要的条件应用过滤器，即 column1 参数不等于 column2 参数，然后应用 reduceByKey

希望我的回答很清楚

谢谢