# Optimization of the Join between Large Tables in the Spark Distributed Framework
## 4 Optimization of Project Analysis  
### Cost Optimization Estimation  
The execution efficiency of Spark Join tasks is affected by the CPU, memory, disk, runtime configuration and execution code. In the process of evaluating the cost of executing tasks, it is difficult for us to calculate the exact cost of tasks. In the case of a fixed configuration, we only need to estimate the cost of Spark Join tasks before and after optimization to obtain a comparative result, so as to reflect the rationality of the optimization scheme. The physical execution plan of the Spark Join task based on cost-based optimization (CBO) is a tree structure, the cost of which is equal to the sum of costs of each execution node, as shown in Figure 6:  
Figure 4: Spark components related to the accumulator. Figure 5: Spark data exchange of broadcast variables. The cost is equal to the sum of the costs of each execution node, where the highest cost is the Join procedure. In the cost estimation formula of CBO, as shown in Formula (1):  
\[Cost=Rows\times Weight+Size\times(1-Weight) \tag{1}\]  
_Rows_ is the number of rows, _Size_ is the size of the data, and _Weight_ is the weight, which is determined by the spark.sql.cbo.joinReorder.card.weight configuration. In Spark Join, when the data in the two tables are not exactly matched, the weight is fixed, and the fact table is pre-filtered using the optimization scheme before joining, then the rows and size are reduced. According to the CBO estimation formula, the cost before optimization is greater than the cost after optimization. The more data you filter, the lower the cost will be. [PERSON] et al. (2018) studied all possible query execution paths in grouping subquery computation overhead and selected effective query execution paths through efficient query algorithms to reduce the cost. Path analysis technology is also applied in all walks of life. [PERSON] et al. (2018) used path analysis technology to test the hypothesis of the dimension of organizational knowledge ability, reveal the various knowledge abilities of the enterprise, and establish the relationship between them. The experimental optimization purpose of this paper is to reduce the Rows and Size before the Shuffle of Join so as to reduce the cost in the maximum cost link. Since the filtering model used in our experiment is very lightweight, it has little impact on the overall cost. After a large amount of irrelevant data have been filtered out in the pre-filtering phase of the experiment, Spark Join tasks may degrade from complex types to simple ones, as shown in Figure 7, greatly reducing the overall cost. Figure 6: Spark execution node cost. Figure 7: Spark Join degradation. This paper studied the cost of Shuffle write and Shuffle read in Shuffle of Spark Join. The cost estimate of Shuffle write workflow is shown in Formula (2):  
\[Cost_{shuffle\ write}=Cost_{cache}+Cost_{sort}+\sum\Big{(}Cost_{buffer}+Cost_{ spill}\Big{)}+Cost_{merge} \tag{2}\]  
\(Cost_{cache}\) represents the cost of reading data into the cache, \(Cost_{sort}\) represents the cost of sorting according to the marked partition, \(Cost_{buffer}+Cost_{ spill}\) represents the cost of each save to the cache and spill to disk and \(Cost_{merge}\) represents the cost of small file consolidation on disk. It should be noted that the operation processes of Shuffle write and Shuffle read are similar, but Shuffle read needs to establish a network connection and data transfer. When the running memory is sufficient, there will be no spill operation, so no disk file will be generated. When the memory is insufficient, it will also generate sort and spill operations to generate disk files, so the cost calculation of Shuffle read is different in the case of sufficient memory and insufficient memory.