scala - 从 scala 中的 map((tuple),(tuple)) 中读取元组的各个元素

Question

reducebykey 生成的输出是一个 ShuffledRDD，key-value 都是多个字段的数组。我需要提取所有字段并写入配置单元表。

下面是我正在尝试的代码:

sqlContext.sql(s"select SUBS_CIRCLE_ID,SUBS_MSISDN,EVENT_START_DT,RMNG_NW_OP_KEY, ACCESS_TYPE  FROM FACT.FCT_MEDIATED_USAGE_DATA")
val USAGE_DATA_Reduce = USAGE_DATA.map{ USAGE_DATA => ((USAGE_DATA.getShort(0), USAGE_DATA.getString(1),USAGE_DATA.getString(2)),
(USAGE_DATA.getInt(3), USAGE_DATA.getInt(4)))}.reduceByKey((x, y) => (math.min(x._1, y._1), math.max(x._2,y._2)))

我期望的最终输出是所有五个字段:SUBS_CIRCLE_ID、SUBS_MSISDN、EVENT_START_DT、MINVAL、MAXVAL

这样就可以直接插入到hive表中

Answer 1

如果你的意思是:

Given a RDD[(TupleN, TupleM)], how do I map each record's elements of both key and value tuples into a single concatenated string?

这是一个简化版本，您应该能够推断它来解决您的问题:

val keyValueRdd = sc.parallelize(Seq(
  (1, "key1") -> (10, "value1", "A"),
  (2, "key2") -> (20, "value2", "B"),
  (3, "key3") -> (30, "value3", "C")
))

val asStrings: RDD[String] = keyValueRdd.map {
  case ((k1, k2), (v1, v2, v3)) => List(k1, k2, v1, v2, v3).mkString(",")
}

asStrings.foreach(println)
// prints:
// 3,key3,30,value3,C
// 2,key2,20,value2,B
// 1,key1,10,value1,A