# Optimization of the Join between Large Tables in the Spark Distributed Framework
## 4 Optimization of Project Analysis
### Optimization of Work Content  
In the Shuffle process of Spark Join, each node of the cluster writes data to the local disk file through Shuffle write, and Shuffle read obtains the disk file of each node through the network transmission. There are a lot of data interactions, network transfers and file read and write operations, which is why the Shuffle phase is very time and resource consuming.  
The optimization scheme in this paper is to preprocess the two tables of Join based on the fact table and dimension table data not completely matching, clean the fact table data before Shuffle, deal with unnecessary data and only let the data that need to be joined enter the Shuffle phase, which saves resources and reduces the running time to a greater extent.  
How to clean the data of each node has become the key to the experiment. Firstly, it should complete the data cleaning task under the condition of limited resources. Secondly, it should have a good cleaning effect in many data sets and should keep the task stable during operation and easy to maintain. According to the requirements of the optimization scheme, the lightweight and highly compressible storage component RoaringBitmap was selected. The accumulator and broadcast variable are used to ensure that RoaringBitmap has high stability, maintainability and efficiency in the process of data loading and data transmission. Therefore, the accumulator, broadcast variable and RoaringBitmap were selected for the Spark pre-filtering task in the experiment.  
The execution flow of Join for the optimization scheme in this paper is shown in Figure 8. We first create an accumulator and load the RoaringBitmap into it, and then collect the dimension table data keys into the accumulator of the type RoaringBitmap. The RoaringBitmap is broadcast to each node as a Spark broadcast variable and stored in memory. In the filtering phase, each cluster node reads the Key stored in the RoaringBitmap and matches the Key of the current fact table. If the Key of the fact table does not match the value, the data will be deleted. Through the above method, a fact table without redundant data is obtained, and then we Join the data. There are no redundant data in the fact table to enter the Shuffle phase, so as to avoid unnecessary data interaction, network transmission and disk reading and writing generated by Shuffle write and Shuffle read. Thus, efficient and energy-efficient Spark Join tasks can be achieved.  
The storage engine in the experiment running task infrastructure in this paper is based on Spark on hive, and the query engine is based on Spark on yarn. We first store the dimension table and fact table data required for the experiment in the Hive [27] database, and then submit the Spark Join job request from the Spark client, which will submit the job  
Figure 8: Experiment to optimize specific execution steps.  
to Yarn [28]. Finally, Yarn reads the dimension and fact tables to be joined from Hive and performs a distributed Spark Join. Figure 9 shows the Spark task execution architecture.  
In this paper, the comparison is mainly based on three aspects: the task running time before and after optimization, the data size written to disk by Shuffle write and the data size read by Shuffle read. The larger the Shuffle write and Shuffle read phases, the higher the disk footprint, the higher the network IO and the higher the disk IO. If the task running time is shortened after optimization, and the data size of Shuffle write to disk and Shuffle read to disk is reduced, the optimization scheme is very feasible in Spark Join tasks.