java-使用Scala Apache Spark合并RDD

我有2个RDD.

RDD1: ((String, String), Int)
RDD2: (String, Int)

例如：

    RDD1

    ((A, X), 1)
    ((B, X), 2)
    ((A, Y), 2)
    ((C, Y), 3)

    RDD2

    (A, 6)
    (B, 7)
    (C, 8)

Output Expected

    ((A, X), 6)
    ((B, X), 14)
    ((A, Y), 12)
    ((C, Y), 24)

在RDD1中,(字符串,字符串)组合是唯一的,而在RDD2中,每个字符串键都是唯一的.
RDD2中的A得分(6)与RDD1的键中具有A的条目的所有得分值相乘.

14 = 7 * 2
12 = 6 * 2
24 = 8 * 3

我写了以下内容,但给我一个关于大小写的错误：

val finalRdd = countRdd.join(countfileRdd).map(case (k, (ls, rs)) => (k, (ls * rs)))

有人可以帮我吗？

解决方法:

您的第一个RDD与第二个RDD的键类型不同(元组(A,X)与A).您应该在加入之前对其进行转换：

val rdd1  = sc.parallelize(List((("A", "X"), 1), (("A", "Y"), 2)))
val rdd2 = sc.parallelize(List(("A", 6)))
val rdd1Transformed = rdd1.map { 
   case ((letter, coord), value) => (letter, (coord, value)) 
}
val result = rdd1Transformed
  .join(rdd2)
  .map { 
    case (letter, ((coord, v1), v2)) => ((letter, coord), v1 * v2) 
  }
result.collect()
res1: Array[((String, String), Int)] = Array(((A,X),6), ((A,Y),12))

标签：scala,apache-spark,java
来源： https://codeday.me/bug/20191120/2043852.html