在spark scala中进行reduceByKey转换后，将所有列取回

Question

我是堆栈溢出和火花的新手。基本上做RDD转换。

我的输入数据：

278222631,2763985,10.02.12,01.01.53,Whatsup,NA,Email,Halter,wagen,28.06.12,313657794,VW,er,i,B,0,23.11.11,234
298106482,2780663,22.02.12,22.02.12,Whatsup,NA,WWW,Halter,wagen,26.06.12,284788860,VW,er,i,B,0,02.06.04,123

我的RDD格式

val dateCov: RDD[(Long, Long, String, String, String, String, String, String, String, String, Long, String, String, String, String, String, String, Long)]

做一些还原Bykey转换映射（[（k， k），（v）]，将coll（1,17）作为key，将coll（18）作为Value。并在还原ByKey上应用了一些函数

例子：

val reducedSortedRDD = dateCov.map(r => { ((r._1, r._11) -> (r._18)) })
      .reduceByKey((x, y) => ((math.min(x, y)))) // find minimum diff
      .map(r => (r._1._1, r._1._2, r._2))
      .sortBy(_._1, true)

1. 我的问题-在reduceByKey函数之后是否可以获取所有其他列，即我的reducedSortedRDD返回类型应该是reducedSortedRDD:RDD[（长、长、字符串、字符串、字符串、字符串、字符串、字符串、字符串、长、字符串、字符串、字符串、字符串、字符串、字符串、字符串）]

与本例不同，不reducedSortedDRDD:RDD[（长、长、长）]。

我使用的是spark 1.4

Answer 1

据我所知，您需要将所有列都放在您的还原ByKey函数中（请记住打乱额外数据的开销），或者您可以将还原SortedRDD与原始数据连接起来。

要使所有列都显示出来，可以执行以下操作：

val reducedSortedRDD = dateCov
  .map(r => ((r._1, r._11),(r._18, r._2, r._3, ..., r._17)))
  .reduceByKey((value1,value2) => if (value1._1 < value2._1) value1 else value2)
  .map{case(key, value) => (key._1, key._2, value._2, value._3, ..., value._17, value._1)}
  .sortBy(_._1, true)

要加入，它看起来像这样：

val keyValuedDateCov = dateCov
  .map(r => ((r._1, r._11, r._18), (r._1, r._2,r._3, ...., r._17)))

val reducedRDD = dateCov
  .map(r => ((r._1, r._11), r._18))
  .reduceByKey((x, y) => math.min(x, y)) // find minimum diff
  .map{case(key, value) => ((key._1, key._2, value), AnyRef)}

val reducedSortedRDD = reducedRDD
  .join(keyValuedDateCov)
  .map{case(key, (_, original)) => (key._1, key._2, original._1, original._2, original._3, ..., original._17, key._3)}
  .sortBy(_._1, true)

联接版本的缺点在于，如果原始数据中的多行在第1列、第17列和第18列中具有完全相同的值，那么最终结果也将包含多个具有这些值的行，因此不能适当地减少。如果保证数据在这些列中不会有多个具有相同值的行，那么应该没有问题。