Spark Scala：如何同时过滤RDD和更新计数器
我的初始RDD是记录类型，记录的布局是：
a_key, b_key,c_key,f_name,l_name,address
现在我必须：
删除具有a_key或b_key或c_key为空/空的记录
我必须同时更新无效记录的计数器。
我这样试过：
sc.register( recordStatsAccumulator, "Stat accumulator for " + filename )
val nullFilteredRecords = records.map{ record =>
if( record.A_KEY.isEmpty ||
record.B_KEY.isEmpty ||
record.C_KEY.isEmpty ){
recordStatsAccumulator.add( ValidationLoggingUtil.INVALID )}
record
}
.filter( record =>
!record.A_KEY.isEmpty &&
!record.B_KEY.isEmpty &&
!record.C_KEY.isEmpty)
但是，这段代码效率不高，因为整个RDD都是两次。首先，更新无效记录的计数器，然后再次删除无效记录。
有没有更好/更有效的方法来做到这一点？
展开
收起
1
条回答
写回答
写回答
-
社区小助手是spark中国社区的管理员，我会定期更新直播回顾等资料和文章干货，还整合了大家在钉群提出的有关spark的问题及回答。我想你可以一步完成这两项操作。像这样：
val nullFilteredRecords = records.filter { record =>
if( record.A_KEY.isEmpty ||record.B_KEY.isEmpty || record.C_KEY.isEmpty ) { recordStatsAccumulator.add( ValidationLoggingUtil.INVALID )}
!record.A_KEY.isEmpty && !record.B_KEY.isEmpty && !record.C_KEY.isEmpty
}2019-07-17 23:23:02赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
相关问答
