# A DAG Refactor Based Automatic Execution Optimization Mechanism for Spark

Hang Zhao, Yu Rao, Donghua Li, Jie Tang, Shaoshan Liu

► To cite this version:

Hang Zhao, Yu Rao, Donghua Li, Jie Tang, Shaoshan Liu. A DAG Refactor Based Automatic Execution Optimization Mechanism for Spark. 16th IFIP International Conference on Network and Parallel Computing (NPC), Aug 2019, Hohhot, China. pp.338-344, 10.1007/978-3-030-30709-7_30. hal-03770537

HAL Id: hal-03770537

https://inria.hal.science/hal-03770537v1

Submitted on 6 Sep 2022

HAL is a multi-disciplinary open access
archive for the deposit and dissemination of sci-
entific research documents, whether they are pub-
lished or not. The documents may come from
teaching and research institutions in France or
abroad, or from public or private research centers.

L'archive ouverte pluridisciplinaire HAL, est
destinée au dépôt et à la diffusion de documents
scientifiques de niveau recherche, publiés ou non,
émanant des établissements d'enseignement et de
recherche français ou étrangers, des laboratoires
publics ou privés.

<figure><img src="image_1.png" /></figure>

Distributed under a Creative Commons CC BY 4.0 - Attribution - International License
<figure><img src="image_2.png" /></figure>

ifip

International Federation for Information Processing

This document is the original author manuscript of a paper submitted to an IFIP
conference proceedings or other IFIP publication by Springer Nature. As such, there
may be some differences in the official published version of the paper. Such
differences, if any, are usually due to reformatting during preparation for publication or
minor corrections made by the author(s) during final proofreading of the publication
manuscript.
A DAG Refactor Based Automatic Execution
Optimization Mechanism For Spark

Hang Zhao¹, Yu Rao¹, Donghua Li¹, Jie Tang¹*, Shaoshan Liu²

¹ South China University of Technology University, Guangzhou 510641, P.R.China

* cstangjie@scut.edu.cn

² Perceptln

**Abstract.** In today's big data era, traditional disk-based MapReduce big data framework encountered bottlenecks due to its lower memory utilization and inefficient orchestration of complex tasks. With the advantage of fully use memory resources, Spark provides a lot of data manipulate operators and use DAG to express the dependences. Spark split entire job to multi-stage according to DAG and schedule them in a distributed execution environment, which better adapted to the new characteristic of big data processing. However, Spark didn't consider the resource requirement of different operators and schedule them indiscriminately, which could cause load imbalances on different nodes in the cluster and cause some node become bottlenecks due to its extraordinary resource consumption. In the past, solve this problem need developers to have a lot of experience of Spark and write code sophisticated. In this paper, we proposed a DAG refactor based automatic execution optimization mechanism for Spark. The experimental results show that the DAG refactor mechanism can greatly improve Spark performance by up to 8.8X without misinterpretation of original program semantics.

**keywords:** Big data, Spark, Semantic DAG, DAG Refactor

# 1 Introduction

With the development of information technology, massive data has been generated every day [1]. Traditional big data processing framework, such as Hadoop, use disk to store intermediate data, always encounter disk I/O bottleneck. Spark use RDD (Resilient Distributed Dataset) to store intermediate data and use Linages to archive fault-tolerate [2], which could archive significant performance improvement compared to Hadoop. Therefore, a lot of applications have implemented in Spark, such as Deep Learning [3], smart city [4, 5], and automatically vehicle.

Spark provides a lot of data manipulate operators and use DAG (Directed Acyclic Graph) to express the dependences, then Spark split entire job to multi-stage according to DAG and schedule them in a distributed execution environment. However, extensive experimentations show that different operators have different running characteristic, while Spark didn't consider the resource requirement of different operators and schedule them indiscriminately. In this paper, we proposed a DAG refactor based automatic execution optimization mechanism for Spark. This mechanism could reconstruct the DAG of original program automatically into another structure with higher execution efficiency. With the automatic DAG refactor, the overall system resource utilization can be effectively improved and task execution time can be greatly reduced.

# 2 DAG in Spark

As shown in Fig. 1, in Spark, *DAGScheduler* divides the Job into several stages according to the wide or narrow depends of RDD. The *DAGScheduler* packages each stage into a *TaskSet* and
hands it over to *TaskScheduler*, which will dispatch task to *Executors*. During scheduling, the *SchedulerBackend* is responsible for providing available resources.

<figure><img src="image_3.png" /><figcaption>**Fig. 1.** The architecture of Spark.</figcaption></figure>

<figure><img src="image_4.png" /><figcaption>**Fig. 2.** Wide vs Narrow Dependencies</figcaption></figure>

Spark provides a rich set of data manipulation operators to build complex processing logic, them can be divided into two categories: (1) Transformation operator, mainly used to describe the conversion relationship between RDD, such as map, filter, and union in the Fig. 2. (2) Action operator, which will trigger Spark to submit job, such as groupByKey and join in Fig. 2.

# 3 Observation on Spark Operator

Spark offers great flexibility to application developers by its rich operators set. However, there still lacks theoretical and experimental research on Spark operators. In this paper, we explore different characteristics of operators in Spark through a large number of experiments and get the observation below: (1) Spark operators can be classified into computation intensive operators and Shuffle intensive operators according to the characteristics of operators. (2) Performance of application varies greatly when different operators contributed to the same semantic. (3) Performance of application varies greatly when execution sequence of operators changes. (4) Data volume decides the execution performance and usage of each operator.

# 4 Automatic DAG Refactor Mechanism

In scheduling, Spark only considers the narrow-dependency or wide-dependency of operators in stages division. It is prone to overlook different resource requirements and runtime feature of operators. Thus resulted schedule decisions can not fully mine the in-memory computing potential. In this paper, we propose a Spark automatic optimization framework based on DAG refactor to take care of such sophisticated work and make execution more efficient automatically.

## 4.1 System Design

The automatic DAG refactor mechanism proposed is shown as Fig. 3. The mechanism can reconstruct DAG by modification of RDD dependency and the user-defined execution function. It mainly includes a general DAG refactor module and an extensible DAG refactor rule library.

<figure><img src="image_5.png" /><figcaption>**Fig. 3.** System architecture diagram</figcaption></figure>
## 4.2 DAG refactor rules library

<table><caption>Table 1. Replacement rules</caption><thead><tr><td>Rules</td><td>Scope of application</td></tr></thead><tbody><tr><td>map -> mapPartitions</td><td>User function overhead too large</td></tr><tr><td>foreach -> foreachPartitions</td><td>User function overhead too large</td></tr><tr><td>groupByKey + map -> reduceByKey</td><td>Shuffle data too large</td></tr><tr><td>groupByKey + mapPartitions -> reduceByKey</td><td>Shuffle data too large</td></tr><tr><td>reduce -> treeReduce/treeAggregate</td><td>Driver side performance bottleneck</td></tr><tr><td>aggregate -> treeAggregate</td><td>Driver side performance bottleneck</td></tr><tr><td>reduce-side join -> map-side join</td><td>Has a broadcastable table</td></tr><tr><td>map + filter -> filter + map</td><td>Data reduction after filter</td></tr><tr><td>filter -> filter + coalesce</td><td>Data skew occurs after Filter</td></tr><tr><td>union + distinct -> distinct + union + distinct</td><td>Very much duplicate data</td></tr></tbody></table>

In this paper, we extracted the characteristics of different operators in Spark running process through a large number of experiments. Then, conclude the replacement rules of Spark operator, and form a Spark operator replacement rule library through the analysis of characteristics of Spark operator and semantic analysis of DAG. All replacement rules are shown in Table 1.

# 5 Implementation in Spark

## 5.1 DAG refactor

The implementation of proposed DAG refactor mechanism mainly by modifying the function *handleJobSubmitted* in the *DAGScheduler* to handle job submission, and the job submitted by the user can be extracted. Then call the *DAGRefactor* component to refactor user job, form a refactored job, and finally replace the original job with refactored job, and then continue to execute by the Spark.

## 5.2 DAGRefactor design and implementation

The DAGRefactor class diagram is shown in Fig. 4. *Origin_job* and *refactored_job* store the original job and refactored job after refactor respectively; *adjacency_table* and *inverseadjacency_table* are intermediate variables of running process, which are used to store the adjacency table and inverse adjacency table of DAG; *rule_list* lists definable refactor rules, it is convenient to add more refactor schemes later.

<figure><img src="image_6.png" /><figcaption>Fig. 4. DAGRefactor class diagram</figcaption></figure>

Depending on their functionality, DAGRefactor provides modules such as DAG Analyse, Rules Match, Refactored Check, DAG Rewrite, and DAG Refactor Rules Library, and provides separate function interfaces for different modules.
# 6 Experiments and Evaluation

## 6.1 Evaluation Environment

By running Spark in real environment, the improvement of DAG refactor strategy proposed in this paper can be evaluated and analyzed. Spark is based on its 2.3.0 version and we set 4 Spark worker instances, each have 4 CPU cores and 6GB memory.

## 6.2 SQL

Spark-SQL is a typical scenario of *foreach* operator. First, we made experiment by using *foreach* operation to inserting 1,000,000 rows and 100,00 rows respectively. Next, we empty the database and proposed framework make refactor by using *foreachPartitions* operator. Table 2 shows experimental results, compared with *foreach* implementation, *foreachPartitions* gives 8.8X speedup at best meanwhile consumes less bandwidth, and no data loss.

<table><caption>Table 2. Comparison of *foreach* and *foreachPartitions*</caption><thead><tr><th></th><th>Executor*core</th><th>Time (s)</th><th>CPU</th><th>Bandwidth</th><th>Data loss</th></tr></thead><tbody><tr><td>foreach (1000K)</td><td>4 * 4</td><td>457.494</td><td>10%</td><td>0~50 Mbps</td><td>79%</td></tr><tr><td>foreachPartitions (1000K)</td><td>4 * 4</td><td>52.089</td><td>5%</td><td>15 Mbps</td><td>0%</td></tr><tr><td>foreach (100K)</td><td>2 * 1</td><td>260.193</td><td>5%</td><td>6 Mbps</td><td>0%</td></tr><tr><td>foreachPartitions (100K)</td><td>2 * 1</td><td>54.054</td><td>5%</td><td>2 Mbps</td><td>0%</td></tr></tbody></table>

## 6.3 Data Aggregation

Data aggregate summarizes all records of RDDs in two phases. As shown in Fig. 5, 30 seconds later, the limited computing resources of driver results in slow execution and diver becomes a performance bottleneck, resulting in total execution time of up to 55 seconds.

Without changing other variables, proposed framework refactor *aggregate* operator into *treeAggregate* operator. As shown in Fig. 6, *treeAggregate* operator adopted a tree-like aggregation strategy in the second phase, thus keeping CPU utilization at a high level consistently. It took only 27 seconds to complete all tasks and reduced execution time by 51%.

<figure><img src="image_7.png" /><figcaption>Fig. 5. *aggregate* resource consumption</figcaption></figure>

<figure><img src="image_8.png" /><figcaption>Fig. 6. *treeAggregate* resource consumption</figcaption></figure>

## 6.4 Merge and Deduplication

We tested the refactor improvement of merging operations and deduplication operations of two RDDs, i.e. *A.union(B).distinct()* is rewritten to *A.distinct().union(B.distinct()).distinct()*. Fig. 7 and Fig. 8 shows the resource consumption respectively. It can be seen that by refactor, total running time of program is reduced from 51s to 47s, and performance is improved by 7.8%.

<figure><img src="image_9.png" /><figcaption>Fig. 7. *union + distinct* resource consumption</figcaption></figure>

<figure><img src="image_10.png" /><figcaption>Fig. 8. *distinct + union + distinct* resource consumption</figcaption></figure>
# 7 Relation Works

Spark provides rich operators and uses them to organize computational logic, but research on Spark operators is still relatively rare. [6] studied the input-output ratio of different operators, to estimate the size of intermediate data in computing process. [7] talked about incremental calculation, studied the difference between different operators when data increments. According to the difficulty of operator multiplexing, its divided operators into two types: indirect multiplexing which can be directly multiplexed and deduced by predicate. At the same time, implemented FQ-Tree-based reusable fragment matching and DAG refactor. For scheduling shuffle class operators, [8] analyzed memory scheduling algorithm in Spark Shuffle phase. Considering the un-balanced memory requirements of different task, fair memory allocation scheduling algorithm can not meet the demand well, proposed an adaptive scheduling algorithm that could dynamically adjust the memory allocation of tasks based on overflow historically.

# 8 Conclusions

In this paper, the different characteristics of different operators in Spark are studied by experiment. With this observation, we design and implement a DAG refactor based automatic execution optimization mechanism for Spark. With a large number of experimental analysis of operators in Spark, we summarize several rules for DAG refactor, which can directly optimize the calculation of related operators. Experiments show that the proposed DAG refactor based automatic execution optimization mechanism can improve Spark performance up to 8.8X by DAG refactor without destroying original program semantics.

# References

1. Pempek T A, Yermolayeva Y A, Calvert S L. College students' social networking experiences on Facebook [J]. Journal of Applied Developmental Psychology, 2009, 30(3):227-238.

2. Zaharia M, Chowdhury M, Das T, et al. Resilient distributed datasets: a fault-tolerant abstraction for in-memory cluster computing[C]// Usenix Conference on Networked Systems Design and Implementation. USENIX Association, 2012:2-2.

3. Hamilton M, Raghunathan S, Matiach I, et al. MMLSpark: Unifying Machine Learning Ecosystems at Massive Scales[J]. arXiv preprint arXiv:1810.08744, 2018.

4. Agafonov A, Yumaganov A. Short-term traffic flow forecasting using a distributed spatial-temporal k nearest neighbors model[C]//2018 IEEE International Conference on Computational Science and Engineering (CSE). IEEE, 2018: 91-98.

5. Nasiri H, Nasehi S, Goudarzi M. A survey of distributed stream processing systems for smart city data analytics[C]//Proceedings of the international conference on smart cities and internet of things. ACM, 2018: 12.

6. Bae J, Jang H, Jin W, et al. Jointly optimizing task granularity and concurrency for in-memory mapreduce frameworks[C]//2017 IEEE International Conference on Big Data (Big Data). IEEE, 2017: 130-140.

7. KanJing. The research of key techniques of incremental computing for DAG-based framework. Beijing University of Technology, 2017.

8. Chen Yingzhi. Analysis and Optimization of Memory Scheduling Algorithm of Spark Shuffle[D]. Zhejiang University, 2016.