{'fraction_non_alphanumeric': 0.023104776136559573, 'fraction_numerical': 0.004425541697798724, 'mean_word_length': 4.813564984964918, 'pattern_counts': {'":': 0, '<': 0, '<?xml version=': 0, '>': 0, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
Load Balancing Mechanism Based on Linear Regression Partition Prediction in Spark

With the rise of cloud computing, the internet of things, social networks and other technologies, information data is expanding rapidly, and traditional processing and storage systems have been difficult to deal with massive data. Spark is a fast and efficient MapReduce implementation developed after Hadoop. However, the shuffle operation in spark will cause the data set on some working nodes to be too large, while other nodes may be idle, which will affect the performance of spark jobs. This phenomenon is called data skew. Aiming at the problem of data skew in spark platform, this paper proposes a load balancing mechanism based on linear regression partition prediction: SP-LRP (Spark load balancing mechanism based on Linear Regression Partition), SP-LRP predicts the partition size of the reduce tasks at run-time, leverages the skew detection algorithm to identify the skew partition and adjusts the task resource allocation according to the fine-grained resource allocation algorithm. We use the benchmark dataset to evaluate SP-LRP, and compare the average execution time of two algorithms and the load balancing degree of reducers in different situations. The experimental confirmed the efficiency of SP-LRP in their respective usage scenarios.


Introduction
With the rapid development of mobile Internet and social networks, the data scale is growing explosively. Nowadays, various e-commerce platforms will generate a large number of hot spot data, which brings about the problem of unbalanced data distribution, that is, data skew. People have entered the era of big data, and the processing of big data is a hot spot of current research and application. Because the normal data distribution theoretically skewed, which is what we call the 20-80 principle: 80% of the wealth is concentrated in the hands of 20% of the people, 80% of the users only use 20% of the functions, 20% of users contribute 80% of hits, large amount of data leads to unbalanced data partition, resulting in data skew. Spark is widely used in search engine, database operation, machine learning, scientific computing and other fields due to its simplicity [1].
Compared with Hadoop [2] and other distributed computing frameworks, Apache Spark[3] provides a more effective implementation mechanism for large-scale data processing. Spark, as a distributed data processing framework, uses the memory computing method to introduce the concept of resilient distributed datasets (RDD) [15], load the data into memory, debase the access delay of data exchange, and achieve the ability of quasi real-time analysis of large datasets. Although Spark's RDD does not divide the job display area into map and reduce tasks like MapReduce [19], Spark also has a stage similar to the shuffle stage in MapReduce, and uses some of MapReduce's terms. The tasks before shuffle are called map tasks, and the tasks after shuffle are called reduce tasks. The main reason for data skew in Spark is its shuffle mechanism. As shown in Figure 1, because Spark has the shuffle mechanism, in the case of data skew, the shuffle operation pulls a large number of data with the same key in different nodes to the same node to perform the reduce operation. A Spark task can only be executed in one partition, so the task running time of some key with huge data volume will be very slow. At present, the main idea of research on Spark task optimization is to avoid shuffle operation and kill data skew in the cradle. The other is to optimize resource allocation in task scheduling. In order to solve the problem of performance and running time, this paper proposes a load balancing mechanism based on linear regression partition prediction, which can predict the partition size of each map task before the completion of map task, improve the load balancing of reduce task, and realize the payload balancing under Spark platform.
The rest of this paper is organized as follows: Section 2 investigates the related work of data skew in distributed processing environment; section 3 introduces a load balancing mechanism based on linear regression partition prediction; section 4 presents experiment results. The conclusion is in Section 5.

Related Works
Nowadays, the framework used to deal with a large number of data is based on MapReduce. The academic community has a more in-depth research and mining on all aspects of it. Data skew is one of the biggest threats to the performance of this system. The concept of Spark appeared in 2012, and the academic research on it is relatively limited. Therefore, to some extent, we can learn from the relevant research and methods in MapReduce to deal with data skew. When solving the data skew problem in map reduce, in order to optimize the performance of MapReduce framework, in recent years, people have conducted a lot of research on the data skew problem in MapReduce, there are also in-depth studies on data skew in distributed databases [20,21]. The solution mainly includes the following aspects: solving the data skew in map stage; improving the data skew in reduce stage, solving the data skew in map stage and reduce stage.
On the map stage, Jiong et al. [4] proposed a mechanism to distribute load in heterogeneous environments according to the capacity of machines. The purpose of this mechanism is to enable map nodes to complete their tasks at the same time. The specific implementation of this mechanism is that when processing files of the same size, the machines in the cluster first sort according to their response time, and then allocate data blocks according to the capacity of the machines on HDFS. After data initialization, if new data is added or deleted, or nodes are removed from the cluster, or new nodes are added to the cluster, the data will be reallocated among nodes according to their computing power.Wang [18] proposed an incremental data allocation approach to reduce partition skew among reducers on MapReduce. The proposed approach divides mapped data into many micro- partitions and gradually gathers the statistics on their sizes in the process of mapping. The micropartitions are then incrementally allocated to reducers in multiple rounds. Guo et al. [5] put forward a strategy to shorten the execution time of MapReduce task, which can shorten the execution time of map phase by better using mapper. In this strategy, if the number of map tasks is less than the number of mappers, many map tasks will be split into several smaller tasks so that all mappers can be used for processing. In another strategy, if the number of map tasks is greater than mapper, multiple map tasks are merged. During execution, if some map tasks last longer than others, the released mapper can be used to process the remaining data to overcome the calculation deviation.
At the reduce stage, Gufler B and others [6] proposed a distributed monitoring system top cluster for capturing data skew in MapReduce system. They solved the problem of estimating the task cost assigned to the reducer according to the given cost model. Xu et al. [7] Proposed two algorithms to overcome the data partition skew of reduce side. In both algorithms, they use a separate MapReduce job running before the original job, whose task is to estimate the frequency of the cluster, and the sampling method is random sampling with uniform distribution. Tang et al. [8] Proposed a segmentation and composition algorithm of skew intermediate data block (SCID), which can improve the load balance of various reduce tasks. Once a data cluster exceeds the remaining capacity of the current bucket, it will be split. SR Ramakrishnan et al. [9] Proposed a static load balancing algorithm, which evenly distributes the work on the reduce of MapReduce jobs, thus significantly reducing the running time. Chen et al. [10] introduced a strategy called MRSIM (reducing reducer skew in MapReduce). According to the strategy, two algorithms are designed: MRSIM-LB and MRSIM-LBF. In the two algorithms in the shuffle phase, the load size on each reducer is recorded by the reducer load monitor, which allows calculating the average load on the reducer. If in the unordered broadcast stage, the reducer will unorder play all part of its data among the remaining reducers, and the unordered play of the reducer has the maximum load stop, then divide the remaining load of the reducer into two equal parts, and distribute between the two reducers.
When dealing with data skew in the reduce and map side, Dhawalia et al. [11] proposed a method to overcome the calculation skew in the map phase and the partition skew in the reduce phase. Ahmad et al. Proposed three methods to optimize the operation of map reduce jobs on heterogeneous clusters: CALB, CAS and PLB.Kwon et al. [12] Introduced skewtune. If the remaining time of the task is more than 1 minute and there are idle nodes in the cluster, the system will dynamically partition the unprocessed data of the task. The scope partition is already used to distribute unprocessed data. Zhang et al. [13] Considered the impact of heterogeneous environment on MapReduce operation. An algorithm named mrheter (to improve the performance of MapReduce in heterogeneous environment) is proposed. The algorithm divides the MapReduce implementation process into two sub phases: mapshuffle and reduce. In this paper, assuming that there is no skew in the data, the reducer is only on the heterogeneous machine. The algorithm tries to load data according to the capacity of the reducer. In the mapping phase, data is provided for each mapper based on its capabilities.

SP-LRP General Framework
In this section, we propose a load balancing mechanism based on linear regression partition prediction, which includes three modules: partition monitor, partition size predictor and resource broker. First, the operation information is counted according to the partition monitor, then the intermediate operation information partition is predicted according to the partition size predictor, and finally the dynamic resource adjustment is carried out by using the resource scheduler.

SP-LRP Summary
The spark job execution process based on SP-LRP is shown in the following figure, with the basic steps as follows: x After starting the map task, the partition monitor in SP-LRP starts to analyze the operation statistics, such as the percentage of partition completed by the map task and the sum of sub partitions generated by the completed map task for reduce task, and the amount of intermediate data generated. x After getting the operation statistics, the size of each sub partition after completing the 100% map task is calculated by the partition predictor, and the current maximum partition is obtained by sorting.
x After predicting the size of each partition, the skew detection algorithm is used to identify the skew partition.
x Shuffle pulls and merges data from mapper nodes. When the resource broker recognizes the skew partition, it arranges the data in the partition in descending order and dynamically adjust the partition.

Partition Monitor
Partition monitor is an important component of SP-LRP. Each worker node sends heartbeat information to master regularly to ensure its availability and update the status of running tasks of a given application. The existing research work focuses on analyzing the information of (key, value) pairs of intermediate results, and predicting the load of reduce tasks through sampling. This type of method will lead to a lot of overhead. Different from the existing research work, this paper extends the heartbeat mechanism during the map task operation.

Monitoring information includes operation information O and Intermediate data
consists of the percentage of processed data in the total data set l D and the total number of sub partitions l j X generated by reduce task j, where l D will change with the progress of spark job.
Therefore, we use l to distinguish the data values measured at different times.
l j X can be expressed in ^¦ j i C i P . C refers to the set of completed map tasks, and P represents the reduce task partition.
The operation statistics j U 'as in equation (1)

Partition Size Predictor
The relationship between the two information in j U is linear.When the map tasks is completed, l j X is the load of reduce task j. Therefore, we use linear regression technology to determine the correlation coefficient between l D and l j X .
Through this formula, we get the maximum load reducer node under the current state, which is recorded as Max (

] [M V RL
), and then we use the skew detection algorithm to determine whether the current reduce node is overloaded, and finally feedback it to the resource manager for dynamic adjustment in real time Wait for the next heartbeat from ResourceManager 13: end if

Data Skew Detection Algorithm
After getting the current maximum partition, in order to better measure the skew degree of data, this paper defines the uniformity degree of data distribution in data set, and puts forward the concept of data skew. The calculation of data inclination refers to the concept of average absolute deviation in classification statistics, counts the times of each key in a data set, and then calculates the average of the absolute value of the deviation of each observation value and arithmetic average value. At the same time, in order to regularize and standardize the results, we introduce the calculation method of relative mean absolute deviation, that is, divide the mean absolute deviation by the arithmetic mean, and finally the definition of data inclination takes the relative mean absolute deviation of one-half. G represents the data inclination, and a represents the number of times each key in the data set appears. The range of data inclination G is from 0 to 1. The closer G is to 1, the greater the data inclination is. The closer G is to 0, the average data distribution is.

Resource Broker
By predicting the partition size statistics, the waiting time in the reduce phase can be minimized. The main idea of resource scheduler is to distribute the identified skew data to the current least loaded reducer. The keys in the partition are sorted in descending order, with the largest key marked as 1 and the rest marked as 0. After the allocation of each cluster is completed, according to the current remaining capacity of all reducers, sort them in descending order again, and repeat the above cluster allocation process. The dynamic allocation of resource broker is shown in algorithm 2. Large 20 Tera-Sort: Sort load is implemented using sortByKey operator. WordCount: The wordcount application counts the number of occurrences of each word in a given file.The map function generates an intermediate tuple (word, 1).Reduce aggregates intermediate tuples from all map tasks and provides a combined output (key, value) pair.The implementation of the WordCount load changed from using the reduceByKey operator to the groupByKey operator, because reduceByKey does map-side local aggregation by default, the amount of data transferred to each Executor is small, and the skew is not obvious.
Application makespan and load balancing degree are often used as the classical metrics [8,16,17] to quantitatively measure the impact of Spark application. There are two main evaluation indexes of the experiment, one is the makespan of the application, the makespan is the total execution time of the benchmark test, that is, the total time consumed from the first task in the first stage of the first job to the last task in the last stage of the last job; the other is the load balancing degree of the cluster, reflecting load balancing degree of spark cluster system.
The experimental comparison algorithm is HashPartitioner algorithm and SCID algorithm [6]. HashPartitioner is the default method used by the spark platform. In this method, the hash value of the key in the middle data is directly used to get the partition ID of the key; SCID first uses the reservoir sampling algorithm to sample, then estimates the size of the data through sampling, identifies the data skew block, and then splits the data skew block to maintain the load balance of the reducer.

Experimental Result
We present next the results of applying the proposed SP-LRP on real Spark workloads so as to evaluate its effectiveness.
In order to address the above partition skew problem, we propose an SP-LRP based on linear regression partition to create containers. In Spark, they used default resource container of uniform specification irrespective the different partition size which can degrade the applications and the  In this experiment, the standard deviation 'as in equation (6 and 7)' of each reducer node load is used to measure the system load balance (LB). The smaller the LB, the more balanced the load of each reducer node. The larger the LB, the more uneven the load of the reducer node.

Conclusion
This paper proposes a load balancing mechanism SP-LRP based on linear regression prediction partition to solve reduce data skew problem in spark platform, builds heterogeneous spark standalone cluster, compare SP-LRP with spark's native partition algorithm HashPartitioner and SCID algorithm in WordCount and Tera-Sort benchmark tests, analysis of SP-LRP performance in different data size;