按用户ID减少输入文件
按用户ID减少输入文件
我正在使用包含userId，seqId，eventType和country的结构化输入文件。我需要通过userId在seqId排序后获取每个字段的最后一个非空值来减少它。对于给定的输入：
userId seqId eventType country
A1600001 2 Update JP
A1600001 3 Update
B2301001 2 Update CH
A1600001 1 Create CH
C1200011 2 Update
C1200011 1 Create IN
减少的结果应该是：
A1600001 3 Update JP
C1200011 2 Update IN
B2301001 2 Update CH
我从以下开始：
scala> val file = sc.textFile("/tmp/sample-events.tsv")
scala> val lines = file.map( x => (x.split("t")(0), x) )
scala> lines.foreach(x => println(x))
(A1600001,A1600001 2 Update JP)
(A1600001,A1600001 3 Update US)
(B2301001,B2301001 2 Update CH)
(A1600001,A1600001 1 Create CH)
(C1200011,C1200011 2 Update RU)
(C1200011,C1200011 1 Create IN)
现在我想减少ByKey行，但我不知道如何构建缩减函数。
写回答
-
社区小助手是spark中国社区的管理员，我会定期更新直播回顾等资料和文章干货，还整合了大家在钉群提出的有关spark的问题及回答。一种可能的方式（假设seqId从不为空）：
准备pair_rdd1通过过滤掉所有空eventType值与映射器，然后再应用reduceByKey重点= userId找到最新的非空eventType每次userId。假设reducer函数需要两[seqId, eventType]对并返回[seqId, eventType]pair，reduce函数应如下所示：(v1 v2) => ( if(v1[seqId] > v2[seqId]) then v1 else v2 )
准备pair_rdd2通过过滤掉所有空country值与映射器，然后再应用reduceByKey重点= userId找到最新的非空country每次userId。假设reducer函数需要两[seqId, country]对并返回[seqId, country]pair，reduce函数应如下所示：(v1 v2) => ( if(v1[seqId] > v2[seqId]) then v1 else v2 )
既然我们也需要最新的seqIdper userId，我们pair_rdd3通过应用reduceByKeykey = userId和reducer函数来准备：(seqId1 seqId2) => max(seqId1, seqId2)
现在我们执行pair_rdd3.leftOuterJoin(pair_rdd1)获取[userId, seqId, eventType]，然后在左连接的结果我们执行.leftOuterJoin(pair_rdd2)到最终获取[userId, seqId, eventType, country]（两个连接都在键= userId）
请注意，我们使用的left join不是inner join此处，因为可能存在所有eventTypes或所有国家/地区为空的用户ID2019-07-17 23:22:57赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
