时间戳

首页 > 解决方案 > Scala当我有两个键时如何使用reduceBykey

scala - Scala当我有两个键时如何使用reduceBykey

问题描述

一行数据格式： 

id: 123456  
Topiclist: ABCDE:1_8;5_10#BCDEF:1_3;7_11

一个 id 可以有很多行：

id: 123456 
Topiclist:ABCDE:1_1;7_2;#BCDEF:1_2;7_11#

目标：(123456, (ABCDE,9,2),(BCDEF,5,2))

主题列表中的记录以 分隔#，ABCDE:1_8;5_10一条记录也是如此。

记录的格式为<topicid>:<topictype>_<topicvalue>

例如对于ABCDE:1_8有

topicid = ABCDE

主题类型 = 1

主题值 = 8

目标：求和 的总和TopicType1，so 的计数频率TopicType1 应为(id, (topicid, value,frequency))，例如：(123456, (ABCDE,9,2),(BCDEF,5,2))

标签： scalamapreduce

解决方案

假设您的数据是“123456!ABCDE:1_8;5_10#BCDEF:1_3;7_11”和“123456!ABCDE:1_1;7_2#BCDEF:1_2;7_11”，所以我们使用“！” 获取您的用户 ID“123456”

rdd.map{f=>
          val userID = f.split("!")(0)
          val items = f.split("!")(1).split("#")
          var result = List[Array[String]]()
          for (item <- items){
            val topicID = item.split(":")(0)
            for (topicTypeValue <- item.split(":")(1).split(";") ){
              println(topicTypeValue);
              if (topicTypeValue.split("_")(0)=="1"){result = result:+Array(topicID,topicTypeValue.split("_")(1),"1") }
            }
          }
          (userID,result)
          }
    .flatMapValues(x=>x).filter(f=>f._2.length==3)
    .map{f=>( (f._1,f._2(0)),(f._2(1).toInt,f._2(2).toInt) )}
    .reduceByKey{case(x,y)=> (x._1+y._1,x._2+y._2) }
    .map(f=>(f._1._1,(f._1._2,f._2._1,f._2._2)))   // (userID, (TopicID,valueSum,frequences) )

输出为 ("12345",("ABCDE",9,2)), ("12345",("BCDEF",5,2)) 与您的输出略有不同，如果您确实需要，可以将此结果分组("12345",("ABCDE",9,2), ("BCDEF",5,2) )

推荐阅读