spark调优(二):UDF减少JOIN和判断
1. 起因
平时写sql语句的时候经常会有大表与小标做关联查询,然后再进行group by等逻辑分组处理,或者是有很多判断条件,sql里有很多if语句,一些区间类的结构查询,这种sql语句直接放到spark上执行,会有大量的shuffle,而且执行时间巨慢
尤其是大表和小标数据差距特别大,大表作为主要处理对象,进行shuffle和map的时候花费大量时间
2. 优化开始
2.1 改成java代码编写程序
首先的一个方法是用java代码编写spark程序,把所有的条件全部打散,或者小表做广播变量,然后每次处理数据时候在进行取值和判断
但这么会让代码可读性比较差,而且如果是用一些工具直接跑sql出计算结果,破坏程序整体性
2.2 使用UDF
UDF(User-Defined Functions)即是用户定义的hive函数。hive自带的函数并不能完全满足业务需求,这时就需要我们自定义函数了
我们这里只做最简单的UDF,就是制作一个hive函数,然后在大表中查询的时候,直接去调用方法把当初需要关联才能获得数据直接返回
首先可以定义一个udf类
public class UDF implements UDF2<Long, Long, Long> { Map<Long, TreeMap<Long, Long>> map; public TripUDF(Broadcast<Map<Long, TreeMap<Long, Long>>> bmap) { this.map = bmap.getValue(); } @Override public Long call(Long id, Long time) throws Exception { if (map.containsKey(terminalId)) { Map.Entry<Long, Long> a = map.get(id).floorEntry(time); Map.Entry<Long, Long> b = map.get(id).ceilingEntry(time); if (null != a && null != b) { if (a.getValue().equals(b.getValue())) { return a.getValue(); } } } return -1L; } }复制代码
这个UDF方法就是先把小表的数据查询出来,做成TreeMap,然后把范围都放进去,广播出去,再每次查询的时候,都用大表到这里去用id和time进行匹配,匹配成功就是要获得的结果
如果用sql去表达,大概就是,大表的time需要去匹配小表的时间段
tablea join tableb on tablea.id=tableb.id and tablea.time >= tableb.timeStart and tablea.time <= tableb.timeEnd复制代码
然后spark去注册UDF方法
伪原创工具 SEO网站优化 https://www.237it.com/
String udfMethod = "structureMap"; spark.udf().register(udfMethod, new UDF(broadcast1), DataTypes.StringType);复制代码
这样直接去查询大表,然后在特定字段使用udf方法,就可以直接获取相应的结果
select id,time,structureMap(id,time) as tag from tablea复制代码
这样tag的最终结果就和直接关联tableb然后再获取其中的值是一样的结果,但具体执行的内容都交给spark去优化
作者:怀瑾握瑜的嘉与嘉
链接:https://juejin.cn/post/7035071227014152200