火花元组获取详细信息/每个键的rdd

Question

我有这样的一排：

(key1,Illinois|111|67342|...)
(key1,Illinois|121|67142|...)
(key2,Hawaii|113|67343|...)
(key1,Illinois|211|67442|...)
(key3,Hawaii|153|66343|...)
(key3,Ohio|193|68343|...)

(1)如何获得唯一的键？

(2)如何获得每个键的行数(key1 -3行、key2 -1行、键3-2行)因此，产出是: 3,1,2)

(3)如何获得每个键(5MB、2MB、3MB)行的字节大小

编辑1.这是我的新代码：

val rdd : RDD[(String, Array[String])] = ...
val rdd_res = rdd.groupByKey().map(row => (row._1, row._2.size, byteSize(row._2)))

val rddKeys = rdd_res.map(row => row._1)
val rddCount = rdd_res.map(row => row._2)     
val rddByteSize = rdd_res.map(row => row._3)

如何实现byteSize？我想要保存到磁盘的大小。

编辑2。

  val rdd_res : RDD[(String, (Int, Int))] = rdd.aggregateByKey((0,0))((accum, value) => (accum._1 + 1, accum._2 + size(value)), (first, second) => (first._1 + second._1, first._2 + second._2))

  val rdd_res_keys = rdd_res.map(row=>row._1).collect().mkString(",")
  val rdd_res_count = rdd_res.map(row=>row._2).collect().map(_._1).mkString(",")
  val rdd_res_bytes = rdd_res.map(row=>row._2).collect().map(_._2).mkString(",")

Answer 1

对于不同的键，您应该切换顺序：

rdd.keys.distinct.collect

但是从技术上来说，这是通过将键数到map...via countByKey中得到的，后者返回key->count的地图。

rdd.countByKey

而且，要获得字节大小，您应该检查this SO question，因为它将依赖于解码。但是，一旦您决定了一个大小方法，那么您就可以通过以下方法获得它：

rdd.aggregateByKey(0)((accum, value) => accum + size(value), _ + _)

或者，你可以一蹴而就：

rdd.aggregateByKey((0,0))((accum, value) => (accum._1 + 1, accum._2 + size(value), (first, second) => (first._1 + second._1, first._2 + second._2))

，这将产生一个RDD[(String, (Int, Int))]，其中元组中的第一个项是键计数，第二个是键大小。

Answer 2

考虑到您有一对RDD (键，值)。

您可以获取密钥并使用下面的方法进行计数

rdd_res = rdd_inp.countByKey

您可以使用下面的方法列出密钥的大小

rdd_size_res = rdd_inp.groupByKey().map((a,b)=>(a,size(b)))

def size(src: List[String]):List[String] = {

    src.map(a =>  (32 + a.length() * 2).toString())


  }

请检查上述方案是否适用于您的方案。