CN110442594A - A kind of Dynamic Execution method towards Spark SQL Aggregation Operators - Google Patents

Abstract

The invention discloses a dynamic execution method for Spark SQL aggregation operators, which is implemented on the Spark system, and aims to make full use of idle resources for idle execution tasks, and improve the execution performance of aggregation operators in the system in the state of data distribution inclination . Its technical solution can be summarized as follows: according to the size of the data partition that the execution task is responsible for processing, two new types of execution tasks are added, namely segment tasks and stealing tasks, and tasks responsible for processing large data partitions are marked as segment tasks. Mark the tasks responsible for processing small data partitions as stealing tasks; segment tasks process data with segments as logical units, and steal tasks actively steal data from segment tasks after processing the allocated small data partitions to process. The dynamic execution method balances the workload among all execution tasks in the state of data skew, improves the processing performance of the aggregation operator as a whole, and is suitable for the execution of the Spark SQL aggregation operator.

Description

A Dynamic Execution Method for Spark SQL Aggregation Operators

technical field

The invention belongs to the technical field of computers, and in particular relates to a dynamic execution method for Spark SQL aggregation operators.

Background technique

With the continuous development of hardware technology, it has led to a continuous decline in memory prices, while memory bandwidth and memory capacity are constantly increasing. The above factors have made the memory computing system represented by Spark gradually become the mainstream big data processing today. platform. Spark provides a library called Spark SQL for processing structured and semi-structured data. Aggregation operators are one of the most frequently occurring and expensive data processing operators. Spark SQL implements a two-stage data aggregation method for the aggregation operator. In the first stage, each node performs an aggregation operation on the local data according to the primary key. In the second stage, the data with the same primary key is transmitted to the same node. Above, after the transmission, each node performs an aggregation operation on the local data, which correspond to two stages (Stage) in Spark.

Caching data in memory in advance and then processing can greatly improve the processing performance of aggregation operators in Spark SQL. However, the performance benefits brought by caching often depend on the data distribution after caching in the actual process. When data distribution is uneven , that is, when data skew occurs, the processing speed of the entire system will be greatly limited, so that the advantages of memory computing cannot be fully utilized. Since the main calculation overhead of the aggregation operator is concentrated in the first stage, the first stage to the second stage There will be a data cleaning operation in the stage, so when data skew occurs, only the execution of the first stage will be affected. In order to avoid data skew, users need to have sufficient prior knowledge of the data to be processed, which is often not possible in real scenarios, which also leads to the common phenomenon of data skew. Most of the existing technologies are based on data sampling to avoid or deal with this problem. In order to avoid data skew, the system needs to sample the data in advance and use the sampled information to evenly divide the data. Because the accuracy of each sampling is difficult to obtain To ensure that it can only reduce the degree of data skew to a certain extent, at the same time, data sampling itself needs to introduce additional performance overhead; when data skew has already occurred, processing data skew often needs to be based on the sampled data. Data is redistributed, which further introduces the overhead of data repartitioning, which deepens performance loss.

To sum up, the cached data distribution has a huge performance impact on the performance of the aggregation operator, mainly resulting in a significant increase in the execution time of the first stage of the aggregation operator. On the one hand, existing optimization schemes introduce sampling overhead to avoid data skew through data sampling, but ensuring the accuracy of data sampling is often a huge challenge, which also makes data distribution often more common in reality. On the other hand, although data skew can be dealt with by redistributing data, redistributing data often results in huge computational and network overhead.

Contents of the invention

The present invention combines the cache partition information provided by the Spark system to realize a lightweight dynamic execution method, with the purpose of optimizing the data processing performance of the first stage of the Spark SQL aggregation operator under data skew.

In order to optimize the performance of the above operators, the technical solution provided by the present invention on the Spark system is a dynamic execution method for SparkSQL aggregation operators. The dynamic execution method includes two stages of data skew detection and task stealing. In the data skew detection stage, a detection algorithm is used to detect the data skew phenomenon: if there is data skew, the partition dependency construction algorithm is executed to identify and build a data partition between large and small data partitions. According to the dependencies, on the basis of the original execution tasks, two new types of execution tasks are added, namely segmentation tasks and stealing tasks, and a big data is assigned to each segmentation task and stealing task Partition and a small data partition; the task stealing stage sets the segmented processing execution method for the segmented task, so that the segmented task processes the allocated large data partition in units of segments; sets the data stealing execution method for the stealing task , so that after the stealing task finishes processing the allocated small data partition, it will actively steal data from the large data partition of the segmentation task and process it.

The detection algorithm in the data inclination detection stage in the described method, its steps are as follows:

Step A-1: Obtain Spark's size information about all cached data partitions from the key-value storage subsystem inside the Spark system;

Step A-2: According to the size information of the data partition, sort all the data partitions from small to large, put them into a new array, mark it as a partitioned array, and set the head and tail pointers to point to the head and tail of the array respectively;

Step A-3: Determine whether the sizes of the partitions pointed to by the head and tail pointers differ by an order of magnitude, if yes, then there is data skew, otherwise, it does not exist;

Step A-4: Run the partition dependency construction algorithm to identify and establish dependencies between large and small data partitions;

Step A-5: According to the dependency relationship, mark the execution tasks responsible for processing large data partitions as segmentation tasks, and mark the execution tasks responsible for processing small data partitions as stealing tasks;

Step A-6: According to the dependencies built in A-4, find the big data partition corresponding to the small data partition that each stealing task is responsible for processing, so that the stealing task can also keep the information of the small data partition that needs to be processed and the information that needs to be stolen. The fetched big data partition information.

The main flow of the partition dependency construction algorithm in the data skew detection stage in the method is as follows:

Step B-1: Determine whether the size of the data partition pointed to by the head and tail pointers of the partition array is an order of magnitude different. If yes, then the partition pointed to by the head pointer is a small data partition, and the partition pointed to by the tail pointer is a large data partition. dependency;

Step B-2: After recording a pair of dependencies, the head pointer is moved backward by one bit, and the tail pointer is moved forward by one bit;

Step B-3: When the size of the partition data pointed to by the head pointer and the tail pointer differs by an order of magnitude, repeat steps B-1 and B-2 until the end when the head and tail pointers are adjacent; when the data pointed to by the head pointer and the tail pointer When the difference in partition size is less than an order of magnitude, go to step B-4;

Step B-4: Keep the position of the tail pointer at this time as K, place the tail pointer at the end of the array, repeat steps B-1 and B-2 to record partition dependencies, and when the tail pointer reaches K, it indicates all dependencies has been constructed.

In the method, the task stealing stage is the segmentation processing execution mode set for the segmentation task, and the process is as follows:

Step C-1: Check whether the allocated large data partition has been built into a cache partition block, if yes, go directly to C-2, if not, then build a cache partition block first, and then go to C-2;

Step C-2: On the basis of the cache partition block, read the data of the corresponding size from the partition data according to the number of reads in the cache partition block each time. After reading, set the data header index of the cache partition block to the original The index value of the data header plus the value after reading the quantity;

Step C-3: Process the read data according to the normal task logic;

Step C-4: Repeat steps C-2 and C-3 until the data head index and data tail index value of all cache partition blocks are the same, and mark the processing flag as 1, indicating that the processing is over.

In the task stealing stage of the method, the data stealing execution mode set for the stealing task is as follows:

Step D-1: Process the allocated small data partitions according to the logic of native execution tasks;

Step D-2: Check whether the large data partition to be stolen has been constructed as a cache partition block, if yes, go directly to D-3, if not, first construct a cache partition block;

Step D-3: Check whether the cache partition block to be stolen has been processed, if yes, return directly, if not, directly enter D-4;

Step D-4: By default, one-third of the unprocessed data is stolen from the cache partition for processing. After the stealing is completed, the data tail index of the cache partition is subtracted from the amount of stolen data.

The cache partition blocks involved in the task stealing phase are generated by the first execution task that accesses a large data partition in the memory in the task stealing phase, and include the following attributes: partition data, data header index, data tail index, The number of reads, processing flags, and cache partition blocks are stored in each executor of the Spark system, and are unique to each execution stage of the Spark system.

When building a cache partition block, first read and save the large data partition in the memory as a partition data. The partition data is essentially an array that stores all the data of the partition; the data head index points to the head of the partition data array by default when it is constructed. part, the default value is 0; the data tail index points to the tail of the partition data array, and the default value is the length value of the array; the number of reads is calculated based on the head index, tail index and the number of segments to be divided, and the default is a large The data partition is divided into ten parts, so the number of reads is the sum of the head index value and the tail index value divided by ten; the processing flag can only be changed by the segmentation task responsible for the partition, the default value is 0, the processing flag The bit is used to notify other tasks accessing the cache partition that all the data in the partition has been processed and executed.

The beneficial effects of the present invention are that, through the above-mentioned dynamic execution method oriented to the Spark SQL aggregation operator, the overhead of sampling and re-partitioning in the traditional scheme is reduced, and the workload between execution tasks is balanced. In the state of data skew, The data processing performance of the aggregation operator has been greatly improved.

Description of drawings

Fig. 1 is the flow chart of the data inclination detection stage of the dynamic execution method facing Spark SQL aggregation operator in the embodiment of the present invention;

Fig. 2 is in the embodiment of the present invention face the dynamic execution method data tilt detection stage of the Spark SQL aggregation operator to the partition dependency construction algorithm flow chart;

Fig. 3 is in the embodiment of the present invention facing the dynamic execution method task of Spark SQL aggregation operator task stealing phase segmentation task execution mode flow chart;

Fig. 4 is a flow chart of the stealing task execution mode in the task stealing stage of the dynamic execution method oriented to the Spark SQL aggregation operator in the embodiment of the present invention.

Detailed ways

The present invention implements a lightweight dynamic execution method on the basis of the Spark system, and its purpose is to optimize the data processing performance of the Spark SQL aggregation operator under data skew. The technical solution of the present invention will be described in detail below in combination with the embodiments and the accompanying drawings.

In the embodiment of the present invention, there is a dynamic execution method for Spark SQL aggregation operators, which modifies three parts of the Spark system, the scheduler (DAGScheduler), the task (Task) and the block manager (BlockManager). The modified scheduler executes the detection algorithm before performing the aggregation operation, checks whether the data is skewed, and if so, executes the partition dependency construction algorithm to identify and build the dependencies between large and small data partitions, and allocates data partitions according to the dependencies perform tasks. There are three types of modified execution tasks, which are newly added segmentation tasks for processing large data partitions and stealing tasks for processing small data partitions in the present invention, and original execution tasks of the Spark system. The size of the data partition processed by the original execution task is moderate, and the original execution mode of Spark is retained, while the newly added segmentation task and stealing task process data in the execution mode of segment processing and data stealing respectively, and modify Subsequent block managers access the data partitions they are responsible for processing.

In the embodiment of the present invention, a dynamic execution method for Spark SQL aggregation operators includes two stages of data skew detection and task stealing. In the data skew detection stage, a detection algorithm is used to detect data skew phenomenon: if data skew is found, execute The partition dependency construction algorithm builds the dependencies between data partitions. According to the dependencies, on the basis of the original execution tasks, two new types of execution tasks are added, namely segmentation tasks and stealing tasks, and each segment A data partition is assigned to the task and the stealing task; in the task stealing stage, the segmented processing execution method is set for the segmented task, so that the segmented task processes the allocated data partition in units of segments; data stealing is set for the stealing task The execution mode enables the stealing task to actively steal data from the data partition of the segmented task and process it after processing the allocated data partition.

The method sets a new data structure for the large data partition in the memory in the block manager, and caches the partition block. It is generated by the first task that accesses a large data partition in the memory during the task stealing phase. Its function is to logically divide a data partition into multiple fine-grained segments, and provide multiple tasks on the same partition. Concurrent execution is guaranteed. The cache partition block contains the following attributes: partition data, data header index, data tail index, read quantity, and processing identification bit, which are stored in the block manager of the Spark system and are unique in each execution stage of the Spark system , which ensures the correctness of concurrent execution of different applications accessing the cache partition.

When building a cache partition block, first read and save the large data partition in the memory as a partition data. The partition data is essentially an array that stores all the data of the partition; the data header index points to the partition data array by default when it is constructed. The head, the default value is 0; the data tail index points to the tail of the partition data array, and the default value is the length value of the array; the number of reads is calculated based on the head index, tail index and the number of segments to be divided, and the default is a The large data partition is divided into ten parts, so the number of reads is the sum of the head index value and the tail index value divided by ten; the processing flag can only be changed by the segmentation task responsible for the partition, the default value is 0, the processing The identification bit is used to notify other tasks accessing the cache partition that all the data in the partition has been processed and executed.

Referring to Fig. 1 for the flowchart of the data inclination detection stage of the present invention, it comprises the following steps:

Step S101: Obtain Spark's size information about all cached data partitions from the key-value storage subsystem inside the Spark system;

Step S102: Arrange the partitions in ascending order according to the size of the buffered data partitions, put them into the array, and set the head and tail pointers to point to the head and tail of the array respectively;

Step S103: Determine whether the sizes of the partitions pointed to by the head and tail pointers differ by an order of magnitude (the default value is 1024), if so, there is data skew, otherwise, it does not exist;

Step S104: If there is data skew, identify large and small data partitions and establish dependencies between them according to the partition dependency construction algorithm;

Step S105: According to the partition dependencies, check the data partition responsible for the execution task. If it is a large data partition, then mark the task as a segmentation task; if it is a small data partition, then mark it as a stealing task; When the partition does not appear in the dependency relationship, the task is a native execution task, and the data partition is processed according to the original execution method of Spark;

Step S106: According to the partition dependency, allocate a large data partition corresponding to the small data partition they are responsible for for the stealing tasks.

Referring to Fig. 2 for the flow chart of partition dependency construction algorithm of the present invention, its main flow is as follows:

Step S201: Determine whether the sizes of the data partitions pointed to by the head and tail pointers of the partition array differ by an order of magnitude (the default value is 1024). If so, the partition pointed to by the head pointer is a small data partition, and the partition pointed to by the tail pointer is a large data partition. retain their dependencies at this point;

Step S202: After recording a pair of dependencies, the head pointer is moved backward by one bit, and the tail pointer is moved forward by one bit;

Step S203: When the size of the partition data pointed to by the head pointer and the tail pointer differs by an order of magnitude, repeat steps S201 and S202 until the end when the head and tail pointers are adjacent; when the size of the partition data pointed to by the head pointer and the tail pointer does not differ by an order of magnitude , enter step S204;

Step S204: Keep the position of the tail pointer at this time as K, place the tail pointer at the end of the array, repeat steps S201 and S202 to record partition dependencies, and when the tail pointer reaches K, it means that all dependencies have been constructed.

Referring to Figure 3, the flow chart of the segmented processing execution mode of segmented task execution in the task stealing stage of the present invention includes the following steps:

Step S301: Check whether the large data partition to be processed has been constructed as a cache partition block, if not, then construct it according to the information of the large data partition: put the data array of the large data partition into the partition data of the cache partition block, and the data header The head of the partition data is placed in the part index, the tail of the partition data is placed in the data tail index, and the default processing flag is marked as 0; if the large data partition has been constructed with a cache partition block, then directly enter S303;

Step S302: According to dividing the sum of data head index and tail index by the number of segments that need to be logically divided (default value is 10), calculate the number of partition data read each time;

Step S303: Read the data from the partition data according to the value of the read quantity and process until all the data are processed, and mark the processing flag as 1 after the processing is completed. Before reading data each time, first lock the cache partition block to avoid mutual interference between multiple execution tasks. After reading the data, set the data header index to the original data header index plus the number of reads The last value indicates the position where the processing will start next time. After the above operations are completed, the lock is released immediately and data processing begins.

The flow chart of the data stealing execution mode of the stealing task execution in the task stealing stage of the present invention is shown in Figure 4, which includes the following steps:

Step S401: read and process the data of the small data partition;

Step S402: Check whether the large data partition to be stolen has been constructed as a cache partition block, if not, then construct it according to the partition information: put the data array of the large data partition into the partition data of the cache partition block, and the data head index Place the head of the partition data, the data tail index places the tail of the partition data, and the default processing flag is marked as 0; if the large data partition has been constructed as a cache partition block, then enter S403;

Step S403: By judging whether the processing flag of the cache partition block is 1, check whether the cache partition block to be stolen has been processed, if yes, then return directly, if not, proceed to the next step;

Step S404: Lock the cache partition to ensure the synchronization of multiple execution tasks; according to the data head index and tail index value when accessing the cache partition, steal part of the unprocessed data from the partition data of the cache partition (default Steal one-third of the unprocessed data, that is, the difference between the tail index value minus the data head index divided by three), and after stealing the data, subtract the value of the data tail index from the amount of stolen data; release the lock and Process the stolen data.

Claims (7)

1. A dynamic execution method for Spark SQL aggregation operators, characterized in that two stages of data skew detection and task stealing are set in the Spark system, and the data skew detection stage uses a detection algorithm to detect the data skew phenomenon: if If there is data skew, the execution partition dependency construction algorithm identifies and builds the dependencies between large and small data partitions. According to the dependencies, two types of new execution tasks are added on the basis of the original execution tasks, namely segmentation tasks and stealing tasks. fetch tasks, and allocate a large data partition and a small data partition for each segmented task and stealing task respectively; the task stealing stage sets the segmented processing execution mode for the segmented task, so that the segmented task takes the segment as the unit Process the allocated large data partition; set the data stealing execution mode for the stealing task, so that after the stealing task has processed the allocated small data partition, it will actively steal data from the large data partition of the segmentation task and process it . 2. The dynamic execution method according to claim 1, characterized in that, the detection algorithm in the data tilt detection stage, its steps are as follows: Step A-1: Obtain Spark's size information about all cached data partitions from the key-value storage subsystem inside the Spark system; Step A-2: According to the size information of the data partition, sort all the data partitions from small to large, put them into a new array, mark it as a partitioned array, and set the head and tail pointers to point to the head and tail of the array respectively; Step A-3: Determine whether the sizes of the partitions pointed to by the head and tail pointers differ by an order of magnitude, if yes, then there is data skew, otherwise, it does not exist; Step A-4: Run the partition dependency construction algorithm to identify and establish dependencies between large and small data partitions; Step A-5: According to the dependency relationship, mark the execution tasks responsible for processing large data partitions as segmentation tasks, and mark the execution tasks responsible for processing small data partitions as stealing tasks; Step A-6: According to the dependencies built in A-4, find the big data partition corresponding to the small data partition that each stealing task is responsible for processing, so that the stealing task can also keep the information of the small data partition that needs to be processed and the information that needs to be stolen. The fetched big data partition information. 3. according to the described dynamic execution method of claim 1 or 2, it is characterized in that, described partition dependency construction algorithm, its steps are as follows: Step B-1: Determine whether the size of the data partition pointed to by the head and tail pointers of the partition array is an order of magnitude different. If yes, then the partition pointed to by the head pointer is a small data partition, and the partition pointed to by the tail pointer is a large data partition. dependency; Step B-2: After recording a pair of dependencies, the head pointer is moved backward by one bit, and the tail pointer is moved forward by one bit; Step B-3: When the size of the partition data pointed to by the head pointer and the tail pointer differs by an order of magnitude, repeat steps B-1 and B-2 until the end when the head and tail pointers are adjacent; when the data pointed to by the head pointer and the tail pointer When the difference in partition size is less than an order of magnitude, go to step B-4; Step B-4: Keep the position of the tail pointer at this time as K, place the tail pointer at the end of the array, repeat steps B-1 and B-2 to record partition dependencies, and when the tail pointer reaches K, it indicates all dependencies has been constructed. 4. The dynamic execution method according to claim 1, wherein said task stealing stage is a segmented processing execution mode provided for segmented tasks, and its steps are as follows: Step C-1: Check whether the allocated large data partition has been built into a cache partition block, if yes, go directly to C-2, if not, then build a cache partition block first, and then go to C-2; Step C-2: On the basis of the cache partition block, read the data of the corresponding size from the partition data according to the number of reads in the cache partition block each time. After reading, set the data header index of the cache partition block to the original The index value of the data header plus the value after reading the quantity; Step C-3: Process the read data according to the normal task logic; Step C-4: Repeat steps C-2 and C-3 until the data head index and data tail index value of all cache partition blocks are the same, and mark the processing flag as 1, indicating that the processing is over. 5. The dynamic execution method according to claim 1, wherein the task stealing stage is a data stealing execution mode set for the stealing task, and the steps are as follows: Step D-1: Process the allocated small data partitions according to the logic of native execution tasks; Step D-2: Check whether the large data partition to be stolen has been constructed as a cache partition block, if yes, go directly to D-3, if not, first construct a cache partition block; Step D-3: Check whether the cache partition block to be stolen has been processed, if yes, return directly, if not, directly enter D-4; Step D-4: By default, one-third of the unprocessed data is stolen from the cache partition for processing. After the stealing is completed, the data tail index of the cache partition is subtracted from the amount of stolen data. 6. The dynamic execution method according to claim 4 or 5, wherein the cache partition block is generated by the first execution task that accesses a large data partition in the memory during the task stealing phase, and includes the following attributes : Partition data, data head index, data tail index, number of reads, processing flags, cache partition blocks are stored in each executor of the Spark system, and are unique in each execution stage of the Spark system. 7. The dynamic execution method according to claim 4 or 5, wherein, in the construction of the cache partition block, at first the large data partition in the memory is read and saved as a partition data, and the partition data is essentially a saved The array of all data in the partition; the data head index points to the head of the partition data array by default during construction, and the default value is 0; the data tail index points to the tail of the partition data array, and the default value is the length of the array; the number of reads depends on the header The part index, the tail index and the number of segments to be divided are calculated together. The default is to divide a large data partition into ten parts, so the number of reads is the sum of the head index value and the tail index value divided by ten; the processing flag is only The segmentation task responsible for the partition can only be changed. The default value is 0. The processing flag is used to notify other tasks accessing the cache partition that all the data in the partition has been processed and executed.