CN114675969B - Elastic scaling stream processing method and system based on self-adaptive load partition - Google Patents

Abstract

The invention discloses an elastic scaling stream processing method based on self-adaptive load partition, which comprises the steps of constructing a stream processing system based on a Flink prototype; constructing a DKG model for distributing data to downstream operator examples and managing computing states in the examples; constructing an index collector model to collect and store performance index data of the stream processing system; sharing performance index data; constructing a discriminator model for calculating an elastic scaling strategy implementation factor and a load partition strategy implementation factor; constructing a corresponding elastic scaling strategy and a load partition strategy; the build reconfiguration controller module applies policies to the stream processing system to complete the flexible scaling stream processing based on the adaptive load partition. The invention also discloses a system for realizing the elastic scaling stream processing method based on the self-adaptive load partition. The invention can realize lower end-to-end processing delay and higher throughput on balanced and inclined data streams, and has high reliability, good implementation effect, science and reasonability.

Description

Elastic scaling stream processing method and system based on adaptive load partitioning

Technical Field

The present invention belongs to the field of computer data processing, and in particular relates to an elastic scaling stream processing method and system based on adaptive load partitioning.

Background Art

Data streams in the real world often change dynamically, which poses a huge challenge to the processing performance of distributed stream processing systems. Under high-speed input, the system will have high processing delays due to insufficient processing performance, and may even cause input data loss. Under low-speed input, a large number of computing resources in the system are idle, and resource utilization is low. Not only that, the skewed distribution of input data streams also exacerbates the uneven resource utilization of distributed stream processing systems, resulting in reduced processing performance.

In order to solve the resource allocation problem in the task execution process of distributed data stream processing systems, researchers have proposed a variety of resource allocation schemes. The static resource allocation scheme is the simplest resource allocation scheme, which implements resource allocation by setting resource configurations that can meet the foreseeable maximum load. The static resource scheme has the characteristics of simple configuration, but it is difficult to predict the maximum workload that may occur in complex environments, and resource allocation needs to be manually adjusted in different environments. Not only that, the static resource configuration scheme does not have load peaks during most of the processing time, which will undoubtedly result in a large amount of computing resource waste.

In order to solve the problem of resource waste in static resource allocation schemes, a large number of dynamic resource allocation schemes have been proposed. Dynamic resource allocation schemes are mainly divided into rule-based methods and model-based methods. Rule-based methods use predefined rules to generate resource configuration schemes: rule-based methods first collect necessary performance indicators, and apply performance indicators to predefined expert rules, judge system runtime symptoms according to the rules and generate resource reconfiguration actions. Since rules need to be designed by domain experts, rule-based methods cannot be universally applied to different production environments. Model-based methods first establish a system model for the distributed stream processing system, and construct an optimization problem of system performance and resource configuration, and then use optimization methods to solve the optimization problem, and use the calculated optimal solution as the next resource allocation scheme. Although model-based methods can produce better scaling effects on balanced data streams, it is difficult to establish an accurate performance model in tilted data streams, resulting in poor results.

Summary of the invention

One of the purposes of the present invention is to provide a scientific and reasonable elastic scaling stream processing method based on adaptive load partitioning with high reliability and good implementation effect.

A second objective of the present invention is to provide a system for implementing the elastic scaling stream processing method based on adaptive load partitioning.

The elastic scaling stream processing method based on adaptive load partitioning provided by the present invention comprises the following steps:

S1. Build a stream processing system based on the Flink prototype using existing technologies;

S2. Based on the stream processing system constructed in step S1, a DKG (Duplicated Key Group) model is constructed to distribute data to downstream operator instances and manage the computing status in the instances;

S3. Build an indicator collector model to collect and store performance indicator data of the stream processing system;

S4. Share the performance indicator data stored in step S3;

S5. construct a discriminator model for calculating elastic scaling strategy implementation factors and load partitioning strategy implementation factors;

S6. According to the elastic scaling strategy implementation factor and the load partitioning strategy implementation factor obtained in step S5, construct the corresponding elastic scaling strategy and load partitioning strategy;

S7. Construct a reconfiguration controller module to apply the strategy obtained in step S6 to the stream processing system, thereby completing elastic scaling stream processing based on adaptive load partitioning.

The construction of the DKG model described in step S2 is used to distribute data to downstream operator instances and manage the computing status in the instances, and specifically includes the following steps:

Calculate the channel for sending the input data according to the load partitioning algorithm, and send the data from the selected channel to the downstream instance;

After receiving the input data, the downstream instance determines whether it contains the set tag:

If the set tag is included, the data is sent directly to the corresponding downstream operator instance, so that the physical partition is applied to the logical partition;

Otherwise, implement logical partitioning according to the hash partitioning method, and then send the data to the downstream operator instance according to the result of the logical partitioning;

Finally, the state of the input data is managed as follows:

When data is transferred to the downstream operator instance, the KG value of the data is calculated using the following formula:

KG=murmurhash(hashcode(key))%P _max

Where murmurhash() is the murmur hash function; hashcode() is the multiplication hash function; P _max is the maximum parallelism supported by the stream processing system; key is the key of the input data; % is the remainder operation;

Then, the storage location SI corresponding to the KG value is calculated using the following formula:

Where N _KG is the maximum number of keyGroups supported by the stream processing system; N _inst is the number of instances; / is a division operation and truncates the integer part;

Then get the status from the local status backend according to the obtained storage location SI:

If the state of the input data does not exist, create a new state in the local state backend;

If the state of the input data exists, directly obtain the corresponding state;

Perform stateful computations and store updated states in the state backend;

Finally, the calculation results of several operators are added together to obtain a unified calculation result, thereby ensuring the correctness of the calculation result.

The construction of the indicator collector model described in step S3 is used to collect and store the performance indicator data of the stream processing system, and specifically includes the following steps:

Each operator instance initializes a local indicator collector to store effective time indicators, processed data volume indicators, output data volume indicators, and time overhead indicators for selecting output channels. When the indicator collector is initialized, the window length and storage path of the indicator persistence process are obtained from the configuration file.

Each time data is processed, the effective time used for processing is calculated: the nanosecond time is recorded before data deserialization, and the nanosecond time is recorded again after data processing is completed and the serialization process is completed. The effective time is obtained by subtracting the two recorded nanosecond times, and the effective time is accumulated to the effective time indicator in the indicator collector;

Update the processing data volume index, output data volume index and time overhead index of the output channel selection process: after deserialization to obtain a data, the processing data volume index increases by 1; when serialization and a data is waiting to be output, the output data volume index increases by 1; the time overhead index of the output channel selection process is calculated in the form of time difference;

After processing a data, at the end of the processing process, determine whether the difference between the recorded nanosecond time and the starting nanosecond time exceeds the configured window length:

If the configured window size is not exceeded, no action is performed;

If the configured window size is exceeded, the indicator calculation and storage operations are performed:

The indicator calculation includes the following formula to calculate the real processing rate and the real output rate:

Where R _true-proc is the true processing rate; N _proc is the amount of processed data; T _useful is the effective time for data processing; R _true-output is the true output rate; N _output is the amount of output data;

After the calculation is completed, the calculation results are stored in the performance indicator data file.

The step S4 of sharing the performance indicator data stored in step S3 specifically includes the following steps:

Use Samba, inotify and mv tools to achieve real-time sharing of performance indicator data files;

Before starting the stream processing system, configure Samba to implement folder sharing and set the storage path of the performance indicator file configured in the inotify monitoring stream processing system.

When the stream processing system stores a performance indicator data file, inotify generates the full path of the performance indicator data file and triggers the mv operation to move the performance indicator data file to the local shared folder;

Samba shares performance indicator data files on multiple host nodes through the SMB protocol, enabling the system to access the performance indicator data of the stream processing system.

The construction of the discriminator model described in step S5 is used to calculate the elastic scaling strategy implementation factor and the load partitioning strategy implementation factor, and specifically includes the following steps:

The elastic scaling policy implementation factor is calculated using the following steps:

Read performance metrics data from a Samba shared file system;

The discriminator reads each performance indicator file and adds the rate information of the operator instances in the file to the topology of the task;

After integrating all performance indicator data according to operators, calculate the real input rate and average real processing rate of each operator; and use the ratio of the real input rate to the real processing rate rounded up as the optimal operator parallelism (i.e., the number of resource allocations);

Compare the optimal operator parallelism with the current operator parallelism, and calculate the elastic scaling strategy implementation factor:

If the total difference between the optimal operator parallelism and the current operator parallelism exceeds the set threshold, the elastic scaling strategy implementation factor is set to the first set value, indicating that the elastic scaling strategy needs to be executed;

If the total difference between the optimal operator parallelism and the current operator parallelism does not exceed the set threshold, the elastic scaling strategy implementation factor is set to the second set value, indicating that the current operator parallelism is not changed;

The load partitioning strategy implementation factor is calculated using the following steps:

Filter out the operator performance indicator data downstream of the logical partition from all the performance indicator data, read the performance indicator data of all operator instances, and then calculate the inverse of the observed processing rate in the performance indicator file as the observed processing time of the operator instance;

Obtain the maximum observed processing time and the minimum observed processing time, and calculate the difference between the two as the maximum waiting time under the load imbalance state;

Use the queuing theory method to calculate the queuing time under load balancing: model each operator instance into a GI/G/1 queuing model, using the formula Estimate the average queuing time of each operator instance, where T _queue is the estimated average queuing time of each operator instance, ρ is the utilization rate, c _a is the arrival time variation coefficient, c _s is the service time variation coefficient, and R _ture-proc is the actual processing rate corresponding to the operator instance;

The difference between the maximum waiting time in the load imbalance state and the average queuing time in the load balance state of each operator is calculated to represent the additional waiting time caused by load imbalance.

Multiply the average utilization of each operator by the additional waiting time to obtain the final additional waiting time determination value.

The time of load partitioning (i.e., the process of selecting output channels) is counted and multiplied by a set coefficient as the final load partitioning time determination value;

Finally, the additional waiting time determination value and the load partition time determination value are sized:

If the additional waiting time determination value is greater than the load partitioning time determination value, the load partitioning strategy implementation factor is set to a third setting value, indicating that the load partitioning strategy needs to be executed;

If the additional waiting time determination value is less than or equal to the load partitioning time determination value, the load partitioning strategy implementation factor is set to a fourth setting value, indicating that the load partitioning strategy does not need to be executed.

The elastic scaling strategy described in step S6 is specifically constructed by adopting the following steps:

Obtain performance indicator data;

Read the directed acyclic graph representing the processing task and store it in the form of an adjacency linked list;

Use the topological sorting algorithm to get the operator order of tasks from the source operator (Source) to the sink operator (Sink);

The expected output rate set by the source operator is used as the input rate of the first downstream operator, and the parallelism P of the operator is calculated as R _true-input is the real input rate; R _true-proc is the real processing rate;

The true output rate R _true-output is calculated as Where N _output is the number of output data of the operator instance, T _useful is the effective time for data processing, and the actual output rate is used as the actual input rate of the downstream operator;

The above calculation process is continuously iterated, from the source operator to the sink operator of the topological structure, and the parallelism of the operators is calculated one by one.

The load partitioning strategy described in step S6 is specifically constructed by adopting the following steps:

Obtain performance indicator data;

Create a fixed-length hash map to store the frequency of different input data, and the hash map only stores possible hot data; at the same time, create an array to store the amount of data sent by different channels;

When a piece of data is input, determine whether the data has been recorded in the hash map:

If the data has been recorded in the hash map, increase the frequency corresponding to the data by 1;

If the data is not recorded in the hash map and the hash map has not reached the upper limit of capacity, the data is recorded in the hash map and the frequency corresponding to the data is recorded as 1;

If the data is not recorded in the hash map and the hash map reaches its capacity limit, the data with the lowest frequency in the data recorded in the hash map is replaced with the current data, and the frequency corresponding to the recorded data is increased by 1;

Each time data is input, the number of input data is updated; when the number of inputs reaches the set value, the frequencies corresponding to all data recorded in the hash map are reduced according to the set ratio;

Each time the frequency is updated, the following steps are used to calculate the current hot data:

Calculate the frequency of all data in the hash map as a percentage of the total input data F _key represents the frequency of data, N _key represents the number of data in the hash map, and N _total represents the total amount of input data;

Using the set parameters, through the formula Calculate the hot data threshold, θ _hot represents the hot data threshold, P represents the parallelism, and θ _def represents the user-defined parameter for adjusting the hot data threshold;

Compare the data frequency and hot data threshold in the hash map: data with a frequency exceeding the hot data threshold is considered hot data;

Judge the input data:

If the input data is hot data, the channel with the least amount of data output among all output channels is selected as the selected output channel;

If the input data is not hot data, the channel with the least data sent out of the two channels is selected as the selected channel;

After selecting a channel, update the amount of data sent by the channel.

The present invention also discloses a system for implementing the elastic scaling stream processing method based on adaptive load partitioning, comprising a Flink system module, a DKG module, an indicator collector module, an indicator file sharing module, a discriminator module, an elastic scaling strategy generation module, a load partitioning strategy generation module and a reconfiguration control module; the Flink system module, the DKG module, the indicator collector module, the indicator file sharing module and the discriminator module are sequentially connected in series; the output end of the discriminator module is simultaneously connected to the input ends of the elastic scaling strategy generation module and the load partitioning strategy generation module; the output ends of the elastic scaling strategy generation module and the load partitioning strategy generation module are simultaneously connected to the reconfiguration control module; the output end of the reconfiguration control module is connected to the Flink system module; the Flink system module is used to construct a stream processing system; the DKG module is used to Build a DKG model, distribute data to downstream operator instances, and manage the computing status in the instances; the indicator collector module is used to build an indicator collector model to collect and store indicator data of the stream processing system; the indicator file sharing module is used to share the stored performance indicator data files; the discriminator module is used to build a discriminator model to calculate the elastic scaling policy implementation factor and the load partitioning policy implementation factor based on the shared performance indicator data; the elastic scaling policy generation module is used to generate the corresponding elastic scaling policy based on the elastic scaling policy implementation factor; the load partitioning strategy generation module is used to generate the corresponding load partitioning strategy based on the load partitioning strategy implementation factor; the reconfiguration control module is used to apply the obtained elastic scaling strategy and load partitioning strategy to the stream processing system, thereby completing elastic scaling stream processing based on adaptive load partitioning.

The elastic scaling stream processing method and system based on adaptive load partitioning provided by the present invention combines load partitioning technology with elastic scaling technology, and also proposes an adaptive load partitioning scheme; therefore, the present invention can not only achieve lower end-to-end processing delay and higher throughput on balanced data streams, but also can achieve lower end-to-end processing delay and higher throughput on tilted data streams; in addition, the present invention has high reliability, good implementation effect and is scientific and reasonable.

BRIEF DESCRIPTION OF THE DRAWINGS

FIG1 is a schematic flow chart of the method of the present invention.

FIG. 2 is a schematic diagram of the processing process of the DKG model in the method of the present invention.

FIG. 3 is a schematic diagram of the processing process of the indicator collector model in the method of the present invention.

FIG. 4 is a schematic diagram of indicator file sharing in the method of the present invention.

FIG5 is a schematic diagram of frequency update in the load partitioning strategy in the method of the present invention.

FIG. 6 is a system function module diagram of the present invention.

DETAILED DESCRIPTION

FIG1 is a schematic diagram of the method flow of the present invention: the elastic scaling flow processing method based on adaptive load partitioning provided by the present invention comprises the following steps:

S1. Build a stream processing system based on the Flink prototype using existing technologies;

S2. Based on the stream processing system constructed in step S1, a DKG (Duplicated Key Group) model is constructed to distribute data to downstream operator instances and manage the computing status in the instances; specifically, the steps are as follows (as shown in FIG2 ):

Calculate the channel for sending the input data according to the load partitioning algorithm, and send the data from the selected channel to the downstream instance;

After receiving the input data, the downstream instance determines whether it contains the set tag:

If the set tag is included, the data is sent directly to the corresponding downstream operator instance, so that the physical partition is applied to the logical partition;

Otherwise, implement logical partitioning according to the hash partitioning method, and then send the data to the downstream operator instance according to the result of the logical partitioning;

Finally, the state of the input data is managed as follows:

When data is transferred to the downstream operator instance, the KG value of the data is calculated using the following formula:

KG=murmurhash(hashcode(key))%P _max

Where murmurhash() is the murmur hash function; hashcode() is the multiplication hash function; P _max is the maximum parallelism supported by the stream processing system; key is the key of the input data; % is the remainder operation;

Then, the storage location SI corresponding to the KG value is calculated using the following formula:

Where N _KG is the maximum number of keyGroups supported by the stream processing system; N _inst is the number of instances; / is a division operation and truncates the integer part;

Then get the status from the local status backend according to the obtained storage location SI:

If the state of the input data does not exist, create a new state in the local state backend;

If the state of the input data exists, directly obtain the corresponding state;

Perform stateful computations and store updated states in the state backend;

Finally, the calculation results of several operators are added together to obtain a unified calculation result, thereby ensuring the correctness of the calculation result;

S3. Build an indicator collector model to collect and store performance indicator data of the stream processing system; specifically, the following steps are included (as shown in Figure 3):

Each operator instance initializes a local indicator collector to store effective time indicators, processed data volume indicators, output data volume indicators, and time overhead indicators for selecting output channels. When the indicator collector is initialized, the window length and storage path of the indicator persistence process are obtained from the configuration file.

Each time data is processed, the effective time used for processing is calculated: the nanosecond time is recorded before data deserialization, and the nanosecond time is recorded again after data processing and serialization is completed. The effective time is obtained by subtracting the two recorded nanosecond times, and the effective time is accumulated to the effective time indicator in the indicator collector; the effective time excludes the time waiting for reading and writing during the calculation process, which can more accurately reflect the processing performance of the system;

Update the processing data volume index, output data volume index and time overhead index of the output channel selection process: after deserialization to obtain a data, the processing data volume index increases by 1; when serialization and a data is waiting to be output, the output data volume index increases by 1; the time overhead index of the output channel selection process is calculated in the form of time difference;

After processing a data, at the end of the processing process, determine whether the difference between the recorded nanosecond time and the starting nanosecond time exceeds the configured window length:

If the configured window size is not exceeded, no action is performed;

If the configured window size is exceeded, the indicator calculation and storage operations are performed:

The indicator calculation includes the following formula to calculate the real processing rate and the real output rate:

Where R _true-proc is the true processing rate; N _proc is the amount of processed data; T _useful is the effective time for data processing; R _true-output is the true output rate; N _output is the amount of output data;

After the calculation is completed, the calculation results are stored in the performance indicator data file;

After completing the design of the DKG model and the indicator collector model, use the Maven tool to compile the Flink source program. After the compilation is complete, deploy the generated Flink system to a distributed environment.

S4. Sharing the performance indicator data stored in step S3; specifically comprising the following steps (as shown in FIG4 ):

Use Samba, inotify and mv tools to achieve real-time sharing of performance indicator data files;

Before starting the stream processing system, configure Samba to implement folder sharing and set the storage path of the performance indicator file configured in the inotify monitoring stream processing system.

When the stream processing system stores a performance indicator data file, inotify generates the full path of the performance indicator data file and triggers the mv operation to move the performance indicator data file to the local shared folder;

Samba shares performance indicator data files on multiple host nodes through the SMB protocol, enabling the system to access the performance indicator data of the stream processing system;

S5. Construct a discriminator model to calculate the elastic scaling strategy implementation factor and the load partitioning strategy implementation factor; specifically, the following steps are included:

The elastic scaling policy implementation factor is calculated using the following steps:

Read performance metrics data from a Samba shared file system;

The discriminator reads each performance indicator file and adds the rate information of the operator instances in the file to the topology of the task;

After integrating all performance indicator data according to operators, calculate the real input rate and average real processing rate of each operator; and use the ratio of the real input rate to the real processing rate rounded up as the optimal operator parallelism (i.e., the number of resource allocations);

Compare the optimal operator parallelism with the current operator parallelism, and calculate the elastic scaling strategy implementation factor:

If the total difference between the optimal operator parallelism and the current operator parallelism exceeds the set threshold, the elastic scaling strategy implementation factor is set to the first set value, indicating that the elastic scaling strategy needs to be executed;

If the total difference between the optimal operator parallelism and the current operator parallelism does not exceed the set threshold, the elastic scaling strategy implementation factor is set to the second set value, indicating that the current operator parallelism is not changed;

Since the load partitioning strategy is applied to each input data and has a larger time overhead than the data processing time; moreover, the load partitioning strategy is not needed in some cases; therefore, the load partitioning strategy implementation factor is calculated using the following steps:

Filter out the operator performance indicator data downstream of the logical partition from all the performance indicator data, read the performance indicator data of all operator instances, and then calculate the inverse of the observed processing rate in the performance indicator file as the observed processing time of the operator instance;

Get the maximum observed processing time and the minimum observed processing time, and calculate the difference between the two as the maximum waiting time under load imbalance; since the actual processing time between different instances of the operator is almost the same, the difference between the observed processing time is manifested as the difference in the time waiting for data input; and if the processing performance is degraded due to load imbalance, there is at least one operator instance in a full load state, and its observed processing time is the smallest; therefore, the maximum waiting time caused by load imbalance between operator instances is represented by the difference between the maximum processing time and the minimum processing time of the instance;

In order to eliminate the influence of the queue time in the balanced state, the queue time in the load balanced state is calculated using the queue theory method: in the ideal load balancing scenario, the input rate and the actual processing rate of all operator instances are the same; therefore, each operator instance is modeled as a GI/G/1 queue model, using the formula Estimate the average queuing time of each operator instance, where T _queue is the estimated average queuing time of each operator instance, ρ is the utilization rate, c _a is the arrival time variation coefficient, c _s is the service time variation coefficient, and R _ture-proc is the actual processing rate corresponding to the operator instance;

The difference between the maximum waiting time in the load imbalance state and the average queuing time in the load balance state of each operator is calculated to represent the additional waiting time caused by load imbalance.

When the load is balanced and the utilization of each operator instance is low, the waiting time calculated by the above method is large. This is because the small error between the operator observation processing rates causes the calculation result to be amplified due to the low instance resource utilization. Therefore, the average utilization of each operator is multiplied by the additional waiting time to obtain the final additional waiting time determination value.

The time of load partitioning (i.e., the process of selecting output channels) is counted and multiplied by a set coefficient as the final load partitioning time determination value;

Finally, the additional waiting time determination value and the load partition time determination value are sized:

If the additional waiting time determination value is greater than the load partitioning time determination value, the load partitioning strategy implementation factor is set to a third setting value, indicating that the load partitioning strategy needs to be executed;

If the additional waiting time determination value is less than or equal to the load partition time determination value, the load partition strategy implementation factor is set to a fourth setting value, indicating that the load partition strategy does not need to be executed;

S6. According to the elastic scaling strategy implementation factor and the load partitioning strategy implementation factor obtained in step S5, construct the corresponding elastic scaling strategy and load partitioning strategy;

In specific implementation, the elastic scaling strategy is constructed by using the following steps:

Obtain performance indicator data;

Read the directed acyclic graph representing the processing task and store it in the form of an adjacency linked list;

Use the topological sorting algorithm to get the operator order of tasks from the source operator (Source) to the sink operator (Sink);

The expected output rate set by the source operator is used as the input rate of the first downstream operator, and the parallelism P of the operator is calculated as R _true-input is the real input rate; R _true-proc is the real processing rate;

The true output rate R _true-output is calculated as Where N _output is the number of output data of the operator instance, T _useful is the effective time for data processing, and the actual output rate is used as the actual input rate of the downstream operator;

The above calculation process is continuously iterated, from the source operator to the sink operator of the topological structure, and the parallelism of the operators is calculated one by one;

The basic idea of load partitioning is to distribute frequently occurring hot data to all downstream operator instances, and infrequent cold data to two fixed operator instances. Candidate instances for infrequent data distribution are generated through two hash functions. Therefore, the load partitioning strategy is constructed by the following steps (as shown in Figure 5):

Obtain performance indicator data;

Create a fixed-length hash map to store the frequency of different input data, and the hash map only stores possible hot data; at the same time, create an array to store the amount of data sent by different channels;

When a piece of data is input, determine whether the data has been recorded in the hash map:

If the data has been recorded in the hash map, increase the frequency corresponding to the data by 1;

If the data is not recorded in the hash map and the hash map has not reached the upper limit of capacity, the data is recorded in the hash map and the frequency corresponding to the data is recorded as 1;

If the data is not recorded in the hash map and the hash map reaches its capacity limit, the data with the lowest frequency in the data recorded in the hash map is replaced with the current data, and the frequency corresponding to the recorded data is increased by 1;

By replacing low-frequency data, possible hot key updates are achieved, so that new hot data does not need a large amount of accumulation in the early stage to become hot data; and low-frequency data and unrecorded data are both cold data and are treated equally in data partitioning; as shown in Figure 5, the data input at the next moment is "two", which is not stored in the hash map; then the element "the" with the lowest frequency is found, and ("two", 2) is used to replace ("the", 1); then the next input data "a" exists in the hash map, and the "a" element item in the hash map is increased by 1;

Most load partitioning methods do not take into account that the popularity of historical data decreases over time, and thus are less sensitive to the current load data distribution. The load partitioning strategy generator uses a method that reduces the frequency of historical data over time. Each time data is input, the number of input data is updated. When the number of inputs reaches the set value, the frequency of all data recorded in the hash map is reduced by a set ratio (for example, multiplied by a coefficient of 0.5).

Each time the frequency is updated, the following steps are used to calculate the current hot data:

Calculate the frequency of all data in the hash map as a percentage of the total input data F _key represents the frequency of data, N _key represents the number of data in the hash map, and N _total represents the total amount of input data;

Using the set parameters, through the formula Calculate the hot data threshold, θ _hot represents the hot data threshold, P represents the parallelism, and θ _def represents the user-defined parameter for adjusting the hot data threshold;

Compare the data frequency and hot data threshold in the hash map: data with a frequency exceeding the hot data threshold is considered hot data;

Judge the input data:

If the input data is hot data, the channel with the least amount of data output among all output channels is selected as the selected output channel;

If the input data is not hot data, the channel with the least data sent out of the two channels is selected as the selected channel;

After selecting a channel, update the amount of data sent by the channel;

S7. Construct a reconfiguration controller module to apply the strategy obtained in step S6 to the stream processing system, thereby completing elastic scaling stream processing based on adaptive load partitioning.

As shown in FIG6 , a system functional module diagram of the present invention is shown: the system disclosed in the present invention for implementing the elastic scaling stream processing method based on adaptive load partitioning comprises a Flink system module, a DKG module, an indicator collector module, an indicator file sharing module, a discriminator module, an elastic scaling strategy generation module, a load partitioning strategy generation module and a reconfiguration control module; the Flink system module, the DKG module, the indicator collector module, the indicator file sharing module and the discriminator module are connected in series in sequence; the output end of the discriminator module is simultaneously connected to the input end of the elastic scaling strategy generation module and the load partitioning strategy generation module; the output end of the elastic scaling strategy generation module and the load partitioning strategy generation module are simultaneously connected to the reconfiguration control module; the output end of the reconfiguration control module is connected to the Flink system module; the Flink system module is used to construct a stream processing system system; the DKG module is used to build a DKG model, distribute data to downstream operator instances, and manage the computing status in the instances; the indicator collector module is used to build an indicator collector model to collect and store indicator data of the stream processing system; the indicator file sharing module is used to share the stored performance indicator data files; the discriminator module is used to build a discriminator model to calculate the elastic scaling policy implementation factor and the load partitioning policy implementation factor based on the shared performance indicator data; the elastic scaling policy generation module is used to generate the corresponding elastic scaling policy based on the elastic scaling policy implementation factor; the load partitioning strategy generation module is used to generate the corresponding load partitioning strategy based on the load partitioning strategy implementation factor; the reconfiguration control module is used to apply the obtained elastic scaling strategy and load partitioning strategy to the stream processing system, thereby completing elastic scaling stream processing based on adaptive load partitioning.

The elastic scaling stream processing method and system based on adaptive load partitioning provided by the present invention can be applied to fields such as network public opinion analysis and industrial sensor data monitoring.

In the scenario of online public opinion analysis, such as counting the popularity of microblog posts, the present invention can not only realize the basic keyword popularity statistics function in low-heat events, but also maintain normal functions in high-heat events without system crashes.

In the industrial sensor data monitoring scenario, the data collected by a large number of industrial sensors are used as stream data, and the stream processing method of the present invention is used to process and analyze the data, so that the stream processing process of the industrial sensor data has the characteristics of low latency and high throughput, ensuring the real-time performance of data processing and being able to detect production problems in a timely manner to avoid losses.

Claims (5)

1. A method for elastically scaling stream processing based on adaptive load partitioning, comprising the following steps: S1. Build a stream processing system based on the Flink prototype using existing technologies; S2. Based on the stream processing system constructed in step S1, a DKG model is constructed to distribute data to downstream operator instances and manage the computing status in the instances; S3. Build an indicator collector model to collect and store performance indicator data of the stream processing system; S4. Share the performance indicator data stored in step S3; S5. Construct a discriminator model to calculate the elastic scaling strategy implementation factor and the load partitioning strategy implementation factor; specifically, the following steps are included: The elastic scaling policy implementation factor is calculated using the following steps: Read performance metrics data from a Samba shared file system; The discriminator reads each performance indicator file and adds the rate information of the operator instances in the file to the topology of the task; After integrating all performance indicator data according to operators, calculate the real input rate and average real processing rate of each operator; and use the ratio of the real input rate to the real processing rate rounded up as the optimal operator parallelism; Compare the optimal operator parallelism with the current operator parallelism, and calculate the elastic scaling strategy implementation factor: If the total difference between the optimal operator parallelism and the current operator parallelism exceeds the set threshold, the elastic scaling strategy implementation factor is set to the first set value, indicating that the elastic scaling strategy needs to be executed; If the total difference between the optimal operator parallelism and the current operator parallelism does not exceed the set threshold, the elastic scaling strategy implementation factor is set to the second set value, indicating that the current operator parallelism is not changed; The load partitioning strategy implementation factor is calculated using the following steps: Filter out the operator performance indicator data downstream of the logical partition from all the performance indicator data, read the performance indicator data of all operator instances, and then calculate the inverse of the observed processing rate in the performance indicator file as the observed processing time of the operator instance; Obtain the maximum observed processing time and the minimum observed processing time, and calculate the difference between the two as the maximum waiting time under the load imbalance state; Use the queuing theory method to calculate the queuing time under load balancing: model each operator instance into a GI/G/1 queuing model, using the formula Estimate the average queuing time of each operator instance, where T _queue is the estimated average queuing time of each operator instance, ρ is the utilization rate, c _a is the arrival time variation coefficient, c _s is the service time variation coefficient, and R _ture-proc is the actual processing rate corresponding to the operator instance; The difference between the maximum waiting time in the load imbalance state and the average queuing time in the load balance state of each operator is calculated to represent the additional waiting time caused by load imbalance. Multiply the average utilization of each operator by the additional waiting time to obtain the final additional waiting time determination value. Count the time of the load partition and multiply it by a set coefficient as the final load partition time determination value; Finally, the additional waiting time determination value and the load partition time determination value are sized: If the additional waiting time determination value is greater than the load partitioning time determination value, the load partitioning strategy implementation factor is set to a third setting value, indicating that the load partitioning strategy needs to be executed; If the additional waiting time determination value is less than or equal to the load partition time determination value, the load partition strategy implementation factor is set to a fourth setting value, indicating that the load partition strategy does not need to be executed; S6. According to the elastic scaling strategy implementation factor and the load partitioning strategy implementation factor obtained in step S5, construct the corresponding elastic scaling strategy and load partitioning strategy; The elastic scaling strategy is specifically constructed by adopting the following steps: Obtain performance indicator data; Read the directed acyclic graph representing the processing task and store it in the form of an adjacency linked list; Use the topological sorting algorithm to get the operator ordering of tasks from source operators to sink operators; The expected output rate set by the source operator is used as the input rate of the first downstream operator, and the parallelism P of the operator is calculated as R _true-input is the real input rate; R _true-proc is the real processing rate; The true output rate R _true-output is calculated as Where N _output is the number of output data of the operator instance, T _useful is the effective time for data processing, and the actual output rate is used as the actual input rate of the downstream operator; The above calculation process is continuously iterated, from the source operator to the sink operator of the topological structure, and the parallelism of the operators is calculated one by one; The load partitioning strategy is specifically constructed by adopting the following steps: Obtain performance indicator data; Create a fixed-length hash map to store the frequency of different input data, and the hash map only stores possible hot data; at the same time, create an array to store the amount of data sent by different channels; When a piece of data is input, determine whether the data has been recorded in the hash map: If the data has been recorded in the hash map, increase the frequency corresponding to the data by 1; If the data is not recorded in the hash map and the hash map has not reached the upper limit of capacity, the data is recorded in the hash map and the frequency corresponding to the data is recorded as 1; If the data is not recorded in the hash map and the hash map reaches its capacity limit, the data with the lowest frequency in the data recorded in the hash map is replaced with the current data, and the frequency corresponding to the recorded data is increased by 1; Each time data is input, the number of input data is updated; when the number of inputs reaches the set value, the frequencies corresponding to all data recorded in the hash map are reduced according to the set ratio; Each time the frequency is updated, the following steps are used to calculate the current hot data: Calculate the frequency of all data in the hash map as a percentage of the total input data F _key represents the frequency of data, N _key represents the number of data in the hash map, and N _total represents the total amount of input data; Using the set parameters, through the formula Calculate the hot data threshold, θ _hot represents the hot data threshold, P represents the parallelism, and θ _def represents the user-defined parameter for adjusting the hot data threshold; Compare the data frequency and hot data threshold in the hash map: data with a frequency exceeding the hot data threshold is considered hot data; Judge the input data: If the input data is hot data, the channel with the least amount of data output among all output channels is selected as the selected output channel; If the input data is not hot data, the channel with the least data sent out of the two channels is selected as the selected channel; After selecting a channel, update the amount of data sent by the channel; S7. Construct a reconfiguration controller module to apply the strategy obtained in step S6 to the stream processing system, thereby completing elastic scaling stream processing based on adaptive load partitioning. 2. According to the elastic scaling stream processing method based on adaptive load partitioning according to claim 1, it is characterized in that the construction of the DKG model in step S2 is used to distribute data to downstream operator instances and manage the computing status in the instances, specifically comprising the following steps: Calculate the channel for sending the input data according to the load partitioning algorithm, and send the data from the selected channel to the downstream instance; After receiving the input data, the downstream instance determines whether it contains the set tag: If the set tag is included, the data is sent directly to the corresponding downstream operator instance, so that the physical partition is applied to the logical partition; Otherwise, implement logical partitioning according to the hash partitioning method, and then send the data to the downstream operator instance according to the result of the logical partitioning; Finally, the state of the input data is managed as follows: When data is transferred to the downstream operator instance, the KG value of the data is calculated using the following formula: KG=murmurhash(hashcode(key))%P _max Where murmurhash() is the murmur hash function; hashcode() is the multiplication hash function; P _max is the maximum parallelism supported by the stream processing system; key is the key of the input data; % is the remainder operation; Then, the storage location SI corresponding to the KG value is calculated using the following formula: Where N _KG is the maximum number of keyGroups supported by the stream processing system; N _inst is the number of instances; / is a division operation and truncates the integer part; Then get the status from the local status backend according to the obtained storage location SI: If the state of the input data does not exist, create a new state in the local state backend; If the state of the input data exists, directly obtain the corresponding state; Perform stateful computations and store updated states in the state backend; Finally, the calculation results of several operators are added together to obtain a unified calculation result, thereby ensuring the correctness of the calculation result. 3. The elastic scaling stream processing method based on adaptive load partitioning according to claim 2 is characterized in that the construction of the indicator collector model in step S3 is used to collect and store performance indicator data of the stream processing system, and specifically includes the following steps: Each operator instance initializes a local indicator collector to store effective time indicators, processed data volume indicators, output data volume indicators, and time overhead indicators for selecting output channels. When the indicator collector is initialized, the window length and storage path of the indicator persistence process are obtained from the configuration file. Each time data is processed, the effective time used for processing is calculated: the nanosecond time is recorded before data deserialization, and the nanosecond time is recorded again after data processing is completed and the serialization process is completed. The effective time is obtained by subtracting the two recorded nanosecond times, and the effective time is accumulated to the effective time indicator in the indicator collector; Update the processing data volume index, output data volume index and time overhead index of the output channel selection process: after deserialization to obtain a data, the processing data volume index increases by 1; when serialization and a data is waiting to be output, the output data volume index increases by 1; the time overhead index of the output channel selection process is calculated in the form of time difference; After processing a data, at the end of the processing process, determine whether the difference between the recorded nanosecond time and the starting nanosecond time exceeds the configured window length: If the configured window size is not exceeded, no action is performed; If the configured window size is exceeded, the indicator calculation and storage operations are performed: The indicator calculation includes the following formula to calculate the real processing rate and the real output rate: Where R _true-proc is the true processing rate; N _proc is the amount of processed data; T _useful is the effective time for data processing; R _true-output is the true output rate; N _output is the amount of output data; After the calculation is completed, the calculation results are stored in the performance indicator data file. 4. The elastic scaling stream processing method based on adaptive load partitioning according to claim 3 is characterized in that the step S4 of sharing the performance indicator data stored in step S3 specifically includes the following steps: Use Samba, inotify and mv tools to achieve real-time sharing of performance indicator data files; Before starting the stream processing system, configure Samba to implement folder sharing and set the storage path of the performance indicator file configured in the inotify monitoring stream processing system. When the stream processing system stores a performance indicator data file, inotify generates the full path of the performance indicator data file and triggers the mv operation to move the performance indicator data file to the local shared folder; Samba shares performance indicator data files on multiple host nodes through the SMB protocol, enabling the system to access the performance indicator data of the stream processing system. 5. A system for implementing the elastic scaling stream processing method based on adaptive load partitioning as described in any one of claims 1 to 4, characterized in that it includes a Flink system module, a DKG module, an indicator collector module, an indicator file sharing module, a discriminator module, an elastic scaling strategy generation module, a load partitioning strategy generation module and a reconfiguration control module; the Flink system module, the DKG module, the indicator collector module, the indicator file sharing module and the discriminator module are connected in series in sequence; the output end of the discriminator module is simultaneously connected to the input end of the elastic scaling strategy generation module and the load partitioning strategy generation module; the output end of the elastic scaling strategy generation module and the load partitioning strategy generation module are simultaneously connected to the reconfiguration control module; the output end of the reconfiguration control module is connected to the Flink system module; the Flink system module is used to build a stream processing system; DKG The G module is used to build the DKG model, distribute data to downstream operator instances, and manage the computing status in the instances; the indicator collector module is used to build the indicator collector model, collect and store the indicator data of the stream processing system; the indicator file sharing module is used to share the stored performance indicator data files; the discriminator module is used to build the discriminator model, and calculate the elastic scaling policy implementation factor and the load partitioning policy implementation factor based on the shared performance indicator data; the elastic scaling policy generation module is used to generate the corresponding elastic scaling policy based on the elastic scaling policy implementation factor; the load partitioning strategy generation module is used to generate the corresponding load partitioning strategy based on the load partitioning strategy implementation factor; the reconfiguration control module is used to apply the obtained elastic scaling strategy and load partitioning strategy to the stream processing system, thereby completing elastic scaling stream processing based on adaptive load partitioning.