Spark Tuning (II): UDF reduces joins and judgments

Hello everyone , I'm Huai Jinyu , A big data sprouts new , There are two gold eaters at home , Jia and Jia , Up can code Lower energy teach My all-around Daddy

If you like my article , Sure [ Focus on ]+[ give the thumbs-up ]+[ Comment on ], Your third company is my driving force , Looking forward to growing with you ~

1.  cause

At ordinary times to write sql Statements often have large tables and small labels for association queries , And then we can move on group by Wait for logical grouping processing , Or there are many judgment conditions ,sql There's a lot in it if sentence , Some interval class structure queries , such sql Statement directly into spark On the implementation , There will be a lot of shuffle, And the execution time is very slow

In particular, the data gap between large tables and small tables is particularly large , Large tables are the main processing objects , Conduct shuffle and map Spend a lot of time

2.  Optimization starts

2.1  Change to java Code writing program

The first method is to use java Code writing spark Program , Break up all the conditions , Or a small table as a broadcast variable , Then, each time we process the data, we take value and judge

But this makes the code less readable , And if you run directly with some tools sql Calculate the result , Destroy program integrity

2.2  Use UDF

UDF（User-Defined Functions） That is, user-defined hive function .hive The built-in functions can not fully meet the business requirements , At this time, we need to customize the function

We only do the simplest UDF, It's about making a hive function , Then when querying in a large table , Directly call the method to directly return the data that can be obtained by association

First of all, you can define a udf class

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;
 }
}


This UDF The method is to query the data of the small table first , Make it TreeMap, Then put the range in , Broadcast out , Every time you query , All use big watches here id and time Match , Matching success is the result you want to get

If you use sql To express , Is probably , Large watch time Need to match the time period of the small table

tablea join tableb 
on tablea.id=tableb.id and 
tablea.time >= tableb.timeStart and 
tablea.time <= tableb.timeEnd


then spark To register UDF Method

String udfMethod = "structureMap";
spark.udf().register(udfMethod, new UDF(broadcast1), DataTypes.StringType);


In this way, you can directly query the large table , Then use... In specific fields udf Method , You can directly obtain the corresponding results

select id,time,structureMap(id,time) as tag from tablea


such tag The end result is directly related to tableb And then get the same result , However, the specific implementation contents are handed over to spark To optimize

Conclusion

If you like my article , Sure [ Focus on ]+[ give the thumbs-up ]+[ Comment on ], Your third company is my driving force , Looking forward to growing with you ~

We can pay attention to the official account 【 Huai Jin holds Yu's Jia and Jia 】, Get the resource download method