CN114780541B - Data partitioning method, device, equipment and medium in micro batch flow processing system - Google Patents

Abstract

The present application relates to the technical field of real-time data stream processing, and provides a data partitioning method, device, computer equipment, storage medium and computer program product in a micro-batch stream processing system. The present application minimizes the time required for preparatory work before batch partitioning through frequency-aware buffering technology, and traverses a balanced binary tree to obtain an ordered list of keys and their frequency-related information, reducing the sorting time in the processing stage. In the batch partitioning stage, the problem is abstracted as a classic packing problem, which limits the degree of key fragmentation, minimizes the cardinality difference between data blocks, and keeps the size of each data block equal, thereby achieving load balancing of data partitions. In the processing stage, the problem is abstracted as a variable capacity packing problem, and the worst-fit algorithm is used to allocate key clusters, ensuring load balancing between tasks, which can greatly improve data processing throughput without increasing latency.

Description

Data partitioning method, device, equipment and medium in micro-batch stream processing system

Technical Field

The present application relates to the technical field of real-time data stream processing, and in particular to a data partitioning method, apparatus, computer equipment, storage medium and computer program product in a micro-batch stream processing system.

Background technique

With the development and maturity of big data technology, the demand for real-time processing has become more and more widespread, often distributed in applications such as social network analysis and click flow analysis. The importance of real-time processing of big data streams is self-evident, which has led to the emergence of a large number of distributed stream processing systems.

In current technologies, some micro-batch stream processing systems, such as Spark Streaming, Comet, and Google Dataflow, use a batch-at-a-time processing model to improve processing throughput. Compared with traditional one-tuple-at-a-time stream processing systems, micro-batch stream processing systems have the advantages of faster speed and more efficient fault tolerance mechanism. However, the micro-batch stream processing systems in this technology use basic data partitioning technology, and their performance is very sensitive to dynamic changes in load characteristics. Resource utilization is very dependent on evenly dividing the workload on the processing unit.

Summary of the invention

Based on this, it is necessary to provide a data partitioning method, apparatus, computer equipment, storage medium and computer program product in a micro-batch stream processing system to address the above technical problems.

In a first aspect, the present application provides a data partitioning method in a micro-batch stream processing system. The method comprises:

Get data stream tuple;

The data stream tuple is maintained based on a hash table and a balanced binary search tree; wherein the hash table stores the key of the data stream tuple, a first pointer pointing to a tuple list corresponding to the key, and a frequency count of the key; the frequency count of the key is also saved to the balanced binary search tree; each key in the hash table has a second pointer pointing to a corresponding frequency count node in the balanced binary search tree;

Traversing the balanced binary search tree to generate the ordered list; wherein the ordered list includes the key, the frequency count of the key and the tuple list corresponding to the key;

Based on the preset partitioning conditions, the data stream tuples in the ordered list are partitioned in batches; wherein each partition is a data block, and each data block stores information on whether the key is split; all data stream tuples sharing the same key value are modeled as a single item, and the preset partitioning conditions include: limiting the number of splits of a single item, minimizing the number of different single items in a data block, and maintaining equal capacity of each data block;

The key clusters are allocated to the buckets of the Reduce stage for processing by using the information of whether the key in the data block is split based on the worst fit algorithm through the Map task; wherein the output of the Map stage is a cluster composed of key values, each key cluster has all data values of the same key; the capacity of the bucket is determined according to the ratio of the number of key clusters to the number of buckets.

In one of the embodiments, the method further includes: recording each batch processing time and batch interval; obtaining the ratio of each batch processing time to the batch interval; according to a preset ratio threshold, obtaining the continuous batch count whose ratio satisfies the preset ratio threshold; and adjusting the Map task and/or Reduce task according to the continuous batch count.

In one of the embodiments, the preset ratio threshold includes a first ratio threshold; obtaining the continuous batch count whose ratio satisfies the preset ratio threshold according to the preset ratio threshold includes: obtaining the continuous batch count whose ratio is greater than the first ratio threshold.

In one of the embodiments, adjusting the Map task and/or Reduce task according to the continuous batch count includes: when the first continuous batch count reaches a preset count threshold, adding Map tasks when the data rate increases, and adding Reduce tasks when the data distribution increases; wherein the first continuous batch count is a continuous batch count whose ratio is greater than the first ratio threshold.

In one of the embodiments, the preset ratio threshold includes a second ratio threshold; obtaining the continuous batch count whose ratio satisfies the preset ratio threshold according to the preset ratio threshold includes: obtaining the continuous batch count whose ratio is less than the second ratio threshold.

In one of the embodiments, adjusting the Map task and/or Reduce task according to the continuous batch count includes: when the second continuous batch count reaches a preset count threshold, reducing the Map task when the data rate decreases, and reducing the Reduce task when the data distribution decreases; wherein the second continuous batch count is a continuous batch count whose ratio is less than the second ratio threshold.

In a second aspect, the present application also provides a dynamic data partitioning device in a micro-batch stream processing system.

The device comprises:

The acquisition module is used to obtain the data stream tuple;

A maintenance module, for maintaining the data stream tuple based on a hash table and a balanced binary search tree; wherein the hash table stores the key of the data stream tuple, a first pointer pointing to a tuple list corresponding to the key, and a frequency count of the key; the frequency count of the key is also saved to the balanced binary search tree; each key in the hash table has a second pointer pointing to a corresponding frequency count node in the balanced binary search tree;

A generating module, configured to traverse the balanced binary search tree to generate the ordered list; wherein the ordered list comprises the key, the frequency count of the key and the tuple list corresponding to the key;

A partitioning module is used to partition the data stream tuples in the ordered list by batches based on preset partitioning conditions; wherein each partition is a data block, and each data block stores information on whether a key is split; all data stream tuples sharing the same key value are modeled as a single item, and the preset partitioning conditions include: limiting the number of splits of a single item, minimizing the number of different single items in a data block, and maintaining equal capacity of each data block;

An allocation processing module is used to allocate key clusters to buckets in the Reduce stage for processing based on the worst fit algorithm using information on whether the keys in the data blocks are split through Map tasks; wherein the output of the Map stage is a cluster composed of key values, each key cluster having all data values of the same key; and the capacity of a bucket is determined according to the ratio of the number of key clusters to the number of buckets.

In a third aspect, the present application further provides a computer device. The computer device includes a memory and a processor, the memory stores a computer program, and the processor implements the following steps when executing the computer program:

Obtain data stream tuples; maintain the data stream tuples based on a hash table and a balanced binary search tree; wherein the hash table stores the key of the data stream tuple, a first pointer to a tuple list corresponding to the key, and a frequency count of the key; the frequency count of the key is also saved to the balanced binary search tree; each key in the hash table has a second pointer to a corresponding frequency count node in the balanced binary search tree; traverse the balanced binary search tree to generate the ordered list; wherein the ordered list contains the key, the frequency count of the key, and the tuple list corresponding to the key; based on a preset partitioning condition, partition the data stream tuples in the ordered list by batches; wherein each partition is a data block, each data block stores information on whether the key is split; all data stream tuples sharing the same key value are modeled as a single item, and the preset partitioning conditions include: limiting the number of split times of a single item, minimizing the number of different single items in a data block, and maintaining the capacity of each data block equal; through a Map task based on a worst-fit algorithm, using the information on whether the key in the data block is split, the key cluster is allocated to the buckets of the Reduce stage for processing; wherein the output of the Map stage is a cluster composed of key values, each key cluster has all data values of the same key; the capacity of a bucket is determined according to the ratio of the number of key clusters to the number of buckets.

In a fourth aspect, the present application further provides a computer-readable storage medium. The computer-readable storage medium stores a computer program, and when the computer program is executed by a processor, the following steps are implemented:

Obtain data stream tuples; maintain the data stream tuples based on a hash table and a balanced binary search tree; wherein the hash table stores the key of the data stream tuple, a first pointer pointing to a tuple list corresponding to the key, and a frequency count of the key; the frequency count of the key is also saved to the balanced binary search tree; each key in the hash table has a second pointer pointing to a corresponding frequency count node in the balanced binary search tree; traverse the balanced binary search tree to generate the ordered list; wherein the ordered list contains the key, the frequency count of the key, and the tuple list corresponding to the key; based on a preset partitioning condition, partition the data stream tuples in the ordered list by batches; wherein each partition is a data block, each data block stores information on whether the key is split; all data stream tuples sharing the same key value are modeled as a single item, and the preset partitioning conditions include: limiting the number of split times of a single item, minimizing the number of different single items in a data block, and maintaining the capacity of each data block equal; the Map task uses the information on whether the key in the data block is split to allocate key clusters to the buckets of the Reduce stage for processing based on the worst fit algorithm; wherein the output of the Map stage is a cluster composed of key values, each key cluster has all data values of the same key; the capacity of a bucket is determined according to the ratio of the number of key clusters to the number of buckets.

In a fifth aspect, the present application further provides a computer program product. The computer program product includes a computer program, and when the computer program is executed by a processor, the following steps are implemented:

Obtain data stream tuples; maintain the data stream tuples based on a hash table and a balanced binary search tree; wherein the hash table stores the key of the data stream tuple, a first pointer pointing to a tuple list corresponding to the key, and a frequency count of the key; the frequency count of the key is also saved to the balanced binary search tree; each key in the hash table has a second pointer pointing to a corresponding frequency count node in the balanced binary search tree; traverse the balanced binary search tree to generate the ordered list; wherein the ordered list contains the key, the frequency count of the key, and the tuple list corresponding to the key; based on a preset partitioning condition, partition the data stream tuples in the ordered list by batches; wherein each partition is a data block, each data block stores information on whether the key is split; all data stream tuples sharing the same key value are modeled as a single item, and the preset partitioning conditions include: limiting the number of split times of a single item, minimizing the number of different single items in a data block, and maintaining the capacity of each data block equal; the Map task uses the information on whether the key in the data block is split to allocate key clusters to the buckets of the Reduce stage for processing based on the worst fit algorithm; wherein the output of the Map stage is a cluster composed of key values, each key cluster has all data values of the same key; the capacity of a bucket is determined according to the ratio of the number of key clusters to the number of buckets.

The data partitioning method, apparatus, computer equipment, storage medium and computer program product in the above-mentioned micro-batch stream processing system minimize the time required for preparation before batch partitioning through frequency-aware buffering technology. Traversing the balanced binary tree can obtain an ordered list of keys and their frequency-related information, reducing the sorting time in the processing stage. In the batch partitioning stage, the problem is abstracted as a classic packing problem, which limits the degree of key fragmentation, minimizes the cardinality difference between data blocks, and keeps the size of each data block equal, thereby achieving load balancing of data partitions. In the processing stage, the problem is abstracted as a variable-capacity packing problem, and the worst-fit algorithm is used to allocate key clusters, thereby ensuring load balancing between tasks, and can greatly improve data processing throughput without increasing latency.

BRIEF DESCRIPTION OF THE DRAWINGS

FIG1 is a schematic flow chart of a data partitioning method in a micro-batch stream processing system according to an embodiment;

FIG2 is a schematic diagram of a process flow of data caching and dynamic partitioning in a micro-batch stream processing system in one embodiment;

FIG3 is a flow chart of a frequency sensing technique in one embodiment;

FIG4 is a flow chart of implementing load balancing partitioning in batch stages in one embodiment;

FIG5 is a flowchart of processing phase partitioning in one embodiment;

FIG6 is a block diagram of a data partitioning device in a micro-batch stream processing system according to an embodiment;

FIG. 7 is a diagram showing the internal structure of a computer device in one embodiment.

Detailed ways

In order to make the purpose, technical solution and advantages of the present application more clearly understood, the present application is further described in detail below in conjunction with the accompanying drawings and embodiments. It should be understood that the specific embodiments described herein are only used to explain the present application and are not used to limit the present application.

The data partitioning method in the micro-batch stream processing system provided in the embodiment of the present application can be applied to a server, which can be implemented by an independent server or a server cluster composed of multiple servers.

The data partitioning method in the micro-batch stream processing system provided by the present application is described in detail below in combination with various embodiments and corresponding drawings.

In one embodiment, as shown in FIG. 1 and in combination with FIG. 2 , a data partitioning method in a micro-batch stream processing system is provided, including:

Step S101, obtaining a data stream tuple;

Step S102, maintaining data stream tuples based on a hash table and a balanced binary search tree; wherein the hash table stores the key of the data stream tuple, a first pointer pointing to a tuple list corresponding to the key, and a frequency count of the key; the frequency count of the key is also saved to the balanced binary search tree; each key in the hash table has a second pointer pointing to a corresponding frequency count node in the balanced binary search tree;

Step S103, traversing the balanced binary search tree to generate an ordered list; wherein the ordered list includes the key, the frequency count of the key and the tuple list corresponding to the key;

The above steps S101 to S103 are mainly to minimize the time required for the preparation work before batch partitioning through frequency-aware buffering technology. Specifically, frequency-aware buffering technology: obtain a batch of data stream tuples, establish a hash table and a balanced binary search tree to maintain the statistical information of the data stream tuples. Specifically, according to the key of the data stream tuple, the data stream tuple is stored in the hash table HTable<k,v _i >, where k is the key, and vi _is the first pointer pointing to the tuple list tupleList _i corresponding to each key. The hash table HTable also stores the frequency count count _i of the key. At the same time, the frequency count count _i of the key will be saved in the balanced binary tree CountTree. Each key in the hash table HTable has a bidirectional pointer (i.e., the second pointer) pointing to the corresponding frequency count node in the balanced binary tree CountTree. The second pointer allows the key count node to be directly updated. Based on this, the balanced binary tree CountTree is traversed to generate an ordered list of keys and their related frequency information <k _i , count _i , tupleList _i >, where k _i represents the i-th key in the data stream tuple. Combined with Figure 3, a more specific process is as follows:

Input data stream S, batch interval is t _start -t _end , set update compensation budget and initial frequency compensation f. First, reset hash table HTable and CountTree used to store frequency counts; then loop through the data stream tuples received within the batch interval, and add 1 to the data stream tuple count Nc; if the key of the data stream tuple is in the hash table HTable, then insert the data stream tuple into the linked list of the key in the hash table HTable, and update the frequency k.Freq _curr of the current key, the difference between the current key frequency and the frequency before update Delta _freq = k.Freq _curr -k.Freq _updated , the difference between the current time and the last update time Delta _time = Time _now -k _{lastUpdateTime} , if the current frequency step k _f.step of the key is equal to Delta _freq or the current time step k _t.step is equal to Delta _time , then update k.Freq _curr , k.budget and k.Freq _updated in CountTree, if k _f.step is equal to Delta _freq , then update If k _t.step is equal to Delta _time , update k _t.step = (t _end - Time _Now ) / k.budget. If the key of the tuple is not in the hash table HTable, increment the count value K of different keys by 1, insert the tuple into the hash table HTable, insert the key of the tuple into the balanced binary tree, initialize k.Freq _curr and k.Freq _updated to 1, initialize k _t.step = (t _end - Time _Now ) / budget, k _f.step = f.

In order to improve the data processing rate, a coarser-grained approach is often used to update, that is, to periodically update the budget times within a certain time interval, where budget is a compensation value determined according to needs. The control parameter fstep is defined, and the node count is updated once every fstep new tuples with the same key are received. Initially, fstep is set to a constant that reflects the optimal step size. Where _Nest is the number of data tuples in the next batch interval at the average data rate, and K _Avg is the average number of different keys in the past batches. Fstep is adaptively updated for each key based on the ratio of the frequency of the current key to the total number of tuples received in the current batch interval, that is, keys that appear more frequently need to receive more data tuples to trigger updates; in order to ensure that all tuple nodes are updated, a time-based control parameter tstep is set to update keys that have not been updated for a long time. This parameter is estimated based on the budget update time consumed by the key and the remaining duration of the batch interval.

Step S104, based on the preset partitioning conditions, the data stream tuples in the ordered list are partitioned by batches; wherein each partition is a data block, and each data block stores information on whether the key is split; all data stream tuples sharing the same key value are modeled as a single item, and the preset partitioning conditions include: limiting the number of splits of a single item, minimizing the number of different single items in a data block, and maintaining the capacity of each data block equal;

The above step S104 is mainly to balance the load partition. Specifically, all data stream tuples sharing the same key value are modeled as a single item, and the data stream tuples in the ordered list in step S103 are partitioned in batches, each partition is a data block, and each data block needs to store information on whether the key is split. The partitioning process needs to meet the following preset partitioning conditions: limit the number of splits of a single item, minimize the number of different items in a data block, and maintain the capacity of each data block equal.

Specifically, step S104 defines the batch partitioning problem as a balanced packing problem for splittable items. Given a set of N different items: k ₁ , k ₂ , …, k _N , each item is of size _Sn , where 1≤n≤N, and given B＝{b ₁ ,b ₂ ,…,b _B } boxes, each box has a capacity of C. Then, the balanced packing problem for splittable items is to allocate each item to different boxes while satisfying the following conditions: (1) for any b _j , j∈[1,B], the number of tuples in the box is equal to the capacity of the box C; (2) for any b _j , j∈[1,B], the number of different items in the box is greater than or equal to N/B; (3) for any item, it is required to be split as few times as possible. When packing, split the data stream tuples corresponding to the keys whose frequency count _i is greater than the ratio of the data block size to the data block cardinality into two items, one of which has a data stream tuple size equal to the ratio of the data block size to the data block cardinality, put it into the data block, and the other is put into a new list; then, assign the remaining keys in the sorted list to the data blocks in a serpentine arrangement, and finally assign the keys in the new list to the data blocks according to the best fit algorithm.

Combined with Figure 4, the more specific process is as follows:

This process can be composed of three independent loop traversal algorithms. Specifically as follows: a) Traverse the above binary tree to obtain the ordered list of keys and their frequency information <k _i , count _i , tupleList _i > and the tuple count value Nc, the count value K of different keys as input, set the required number of data partitions P; define the partition size P _Size = Nc/P, the partition cardinality P _k = K/P, the threshold for splitting keys S _cut = P _Size / P _k , and set the current partition b _j to be the first partition b ₁ ; b) Traverse the keys in the list, when its count _i is greater than S _cut , put S _cut tuples into b _j , and put the remaining part into the temporary list RList, and update the position of b _j corresponding to the key to Pos(k) = b _j ; and set b _j = b _j%P , j increases by 1, and then repeat step b) until there is no count _i greater than S _cut ; c) Traverse the remaining keys in List, traverse partition bj and put a key in turn, reverse the partition order after traversing the partitions, and repeat step c); d) Traverse the keys in the temporary list RList, set b = Pos(k), if the key can be fully placed in b, put the key in b, otherwise fill b first, and then put the remaining part into the partition with the smallest remaining capacity that can accommodate it.

Step S105, using the Map task based on the worst fit algorithm and using the information of whether the keys in the data block are split, the key clusters are allocated to the buckets of the Reduce stage for processing; wherein the output of the Map stage is a cluster composed of key values, each key cluster has all data values of the same key; the capacity of the bucket is determined according to the ratio of the number of key clusters to the number of buckets.

Step S105 is the processing stage partitioning. In the load balancing batch partitioning step, each data block has information about whether the key is split. The Map task is used to process the database. The Map task uses the information that the key is split to allocate key clusters to the buckets of the Reduce stage for processing. Among them, the output of the Map stage is a cluster composed of key-value pairs. Each key cluster has all data values of the same key. The key cluster C _k can be expressed as C _k = {(k,v _i )|v _i ∈ k}, where vi is the data value corresponding to key k; assume that the K key clusters output by a given Map stage need to be allocated to r Reduce buckets. The output of the Map task is I = {C _k |k∈K}. In order to ensure the load balance of the Reduce stage, it is necessary to ensure that the allocation of each Reduce bucket is consistent, so the capacity of the bucket is set to The partitioning problem in the processing phase can be simplified to a bin packing problem; consider key clusters as items and Reducebuckets as bins. Unlike the batch partitioning problem, the partitioning problem in the processing phase is a variable capacity balanced bin packing problem, which is defined as follows: given a set with M items, A bins a ₁ , a ₂ , …, a _A , each with a capacity C _i , then the variable capacity balanced bin packing problem is to distribute items to different bins a ₁ , a ₂ , …, a _A under the following conditions: (1) For any a _j , j∈[1,A], the number of tuples in the bin is less than the capacity C _j of the bin; (2) For any a _j , j∈[1,A], the number of different items in the bin is greater than or equal to M. Combined with Figure 5, the more specific process is as follows:

As shown in Figure 2, the partition results will enter the Map task for processing. Figure 5 shows the detailed process of assigning the intermediate results of the Map task to the Reduce buckets. First, the input information is the key cluster C obtained by the Map task. The data partition obtained through the above steps contains information on whether the key has been split. The set R of all buckets in the Reduce stage sets Bucksize = |C|/|R|. Use the Hash algorithm to assign the split keys so that only the unsplit keys remain in the key cluster, and sort them in descending order. Then traverse the keys in the key cluster, and assign the larger key cluster to the rth bucket as much as possible according to the worst fit algorithm, and delete the rth bucket from R. If there is no bucket in R, reset R to all buckets, and continue to traverse the remaining keys in the key cluster.

The data partitioning method in the above-mentioned micro-batch stream processing system minimizes the time required for preparation before batch partitioning through frequency-aware buffering technology. Traversing the balanced binary tree can obtain an ordered list of keys and their frequency-related information, reducing the sorting time in the processing stage. In the batch partitioning stage, the problem is abstracted as a classic packing problem, which limits the degree of key fragmentation, minimizes the cardinality difference between data blocks, and keeps the size of each data block equal, thereby achieving load balancing of data partitions. In the processing stage, the problem is abstracted as a variable-capacity packing problem, and the worst-fit algorithm is used to allocate key clusters, which ensures load balancing between tasks and can greatly improve data processing throughput without increasing latency.

In some embodiments, the above method may further include the following steps:

Record each batch processing time and batch interval; obtain the ratio of each batch processing time to the batch interval; according to a preset ratio threshold, obtain the continuous batch count whose ratio meets the preset ratio threshold; according to the continuous batch count, adjust the Map task and/or Reduce task.

This embodiment mainly manages resources dynamically. It changes the degree of parallelism at runtime by setting a threshold for Map-Reduce task processing time, and adjusts Map-Reduce tasks according to changes in workload. Specifically, it continuously adjusts Map tasks and/or Reduce tasks based on the ratio of each batch processing time to the time interval between two batches of data stream tuples.

In some embodiments, the preset ratio threshold includes a first ratio threshold; the above-mentioned obtaining the continuous batch count whose ratio satisfies the preset ratio threshold according to the preset ratio threshold includes: obtaining the continuous batch count whose ratio is greater than the first ratio threshold.

Furthermore, the above adjustment of Map tasks and/or Reduce tasks according to the continuous batch count specifically includes:

When the first continuous batch count reaches a preset count threshold, Map tasks are added when the data rate increases, and Reduce tasks are added when the data distribution increases; wherein the first continuous batch count is a continuous batch count whose ratio is greater than a first ratio threshold.

In some other embodiments, the preset ratio threshold includes a second ratio threshold; the above-mentioned obtaining the continuous batch count whose ratio meets the preset ratio threshold according to the preset ratio threshold includes: obtaining the continuous batch count whose ratio is less than the second ratio threshold.

Furthermore, the above adjustment of Map tasks and/or Reduce tasks according to the continuous batch count specifically includes:

When the second continuous batch count reaches a preset count threshold, the Map task is reduced when the data rate decreases, and the Reduce task is reduced when the data distribution decreases; wherein the second continuous batch count is a continuous batch count whose ratio is less than the second ratio threshold.

In the above embodiment, when the ratio of the processing time of each batch to the time interval between two batches of data stream tuples exceeds or falls below the set threshold for several consecutive batches, the adjustment of the Map-Reduce task is triggered. The details are as follows:

Use Stats _d to record the ratio of the processing time of the first d batches (preset count threshold) to the batch interval, as well as status information such as data rate and data distribution, and define the ratio of batch processing time to batch interval Each batch adds the ratio, data rate and data distribution to Stats _d . Set the first ratio threshold to thres ₁ , and use count to represent the count of several consecutive batches with _Wi > thres ₁ , that is, the first consecutive batch count. If _Wi < thres ₁ occurs, the first consecutive batch count can be reset to zero and recounted. When the first consecutive batch count is equal to d, that is, _Wi for d consecutive batches is greater than the preset count threshold, if the data rate increases, the corresponding Map task is added, and if the data distribution increases, the Reduce task is added; similarly, set the second ratio threshold to thres ₂ , when _Wi < thres ₂ for d consecutive batches, the corresponding tasks are reduced according to the changes in the data rate and data distribution, that is, if the data rate decreases, the Map task is reduced, and if the data distribution decreases, the Reduce task is reduced.

The above embodiment uses dynamic resource management technology to achieve dynamic load adjustment and adjust the degree of parallelism during runtime, so that the method is robust to fluctuations in data distribution and arrival rate, and can significantly improve data processing throughput without increasing latency.

It should be understood that, although the various steps in the flowcharts involved in the above-mentioned embodiments are displayed in sequence according to the indication of the arrows, these steps are not necessarily executed in sequence according to the order indicated by the arrows. Unless there is a clear explanation in this article, the execution of these steps does not have a strict order restriction, and these steps can be executed in other orders. Moreover, at least a part of the steps in the flowcharts involved in the above-mentioned embodiments can include multiple steps or multiple stages, and these steps or stages are not necessarily executed at the same time, but can be executed at different times, and the execution order of these steps or stages is not necessarily carried out in sequence, but can be executed in turn or alternately with other steps or at least a part of the steps or stages in other steps.

Based on the same inventive concept, the embodiment of the present application also provides a data partitioning device in a micro-batch stream processing system for implementing the data partitioning method in the micro-batch stream processing system involved above. The implementation scheme for solving the problem provided by the device is similar to the implementation scheme recorded in the above method, so the specific limitations in the embodiments of the data partitioning device in one or more micro-batch stream processing systems provided below can refer to the limitations of the data partitioning method in the micro-batch stream processing system above, and will not be repeated here.

In one embodiment, as shown in FIG6 , a data partitioning device in a micro-batch stream processing system is provided, and the device 600 may include:

An acquisition module 601 is used to acquire a data stream tuple;

A maintenance module 602 is used to maintain the data stream tuple based on a hash table and a balanced binary search tree; wherein the hash table stores the key of the data stream tuple, a first pointer pointing to a tuple list corresponding to the key, and a frequency count of the key; the frequency count of the key is also saved to the balanced binary search tree; each key in the hash table has a second pointer pointing to a corresponding frequency count node in the balanced binary search tree;

A generating module 603 is used to traverse the balanced binary search tree to generate the ordered list; wherein the ordered list includes the key, the frequency count of the key and the tuple list corresponding to the key;

A partitioning module 604 is used to partition the data stream tuples in the ordered list by batches based on a preset partitioning condition; wherein each partition is a data block, and each data block stores information on whether the key is split; all data stream tuples sharing the same key value are modeled as a single item, and the preset partitioning condition includes: limiting the number of splits of a single item, minimizing the number of different single items in a data block, and maintaining equal capacity of each data block;

The allocation processing module 605 is used to allocate key clusters to the buckets of the Reduce stage for processing based on the worst fit algorithm using the information of whether the key in the data block is split through the Map task; wherein the output of the Map stage is a cluster composed of key values, each key cluster has all data values of the same key; the capacity of the bucket is determined according to the ratio of the number of key clusters to the number of buckets.

In one embodiment, the apparatus 600 may further include:

The task adjustment module is used to record each batch processing time and batch interval; obtain the ratio of each batch processing time to the batch interval; according to a preset ratio threshold, obtain the continuous batch count whose ratio meets the preset ratio threshold; according to the continuous batch count, adjust the Map task and/or Reduce task.

In one embodiment, the preset ratio threshold comprises a first ratio threshold; and the task adjustment module is used to obtain a count of consecutive batches whose ratio is greater than the first ratio threshold.

In one embodiment, a task adjustment module is used to increase Map tasks when the data rate increases, and to increase Reduce tasks when the data distribution increases, when the first continuous batch count reaches a preset count threshold; wherein the first continuous batch count is a continuous batch count whose ratio is greater than the first ratio threshold.

In one embodiment, the preset ratio threshold includes a second ratio threshold; and the task adjustment module is used to obtain a continuous batch count in which the ratio is less than the second ratio threshold.

In one embodiment, a task adjustment module is used to reduce Map tasks when the data rate decreases, and to reduce Reduce tasks when the data distribution decreases, when the second continuous batch count reaches a preset count threshold; wherein the second continuous batch count is a continuous batch count whose ratio is less than the second ratio threshold.

Each module in the data partitioning device in the micro-batch stream processing system can be implemented in whole or in part by software, hardware, or a combination thereof. Each module can be embedded in or independent of a processor in a computer device in the form of hardware, or can be stored in a memory in a computer device in the form of software, so that the processor can call and execute operations corresponding to each module.

In one embodiment, a computer device is provided, which may be a server, and its internal structure diagram may be shown in FIG7. The computer device includes a processor, a memory, and a network interface connected via a system bus. The processor of the computer device is used to provide computing and control capabilities. The memory of the computer device includes a non-volatile storage medium and an internal memory. The non-volatile storage medium stores an operating system, a computer program, and a database. The internal memory provides an environment for the operation of the operating system and the computer program in the non-volatile storage medium. The database of the computer device is used to store data such as data stream tuples. The network interface of the computer device is used to communicate with an external terminal via a network connection. When the computer program is executed by the processor, a data partitioning method in a micro-batch stream processing system is implemented.

Those skilled in the art will understand that the structure shown in FIG. 7 is merely a block diagram of a partial structure related to the solution of the present application, and does not constitute a limitation on the computer device to which the solution of the present application is applied. The specific computer device may include more or fewer components than shown in the figure, or combine certain components, or have a different arrangement of components.

In one embodiment, a computer device is further provided, including a memory and a processor, wherein a computer program is stored in the memory, and the processor implements the steps in the above method embodiments when executing the computer program.

In one embodiment, a computer-readable storage medium is provided, on which a computer program is stored. When the computer program is executed by a processor, the steps in the above-mentioned method embodiments are implemented.

In one embodiment, a computer program product is provided, including a computer program, which implements the steps in the above method embodiments when executed by a processor.

Those of ordinary skill in the art can understand that all or part of the processes in the above-mentioned embodiment methods can be completed by instructing the relevant hardware through a computer program, and the computer program can be stored in a non-volatile computer-readable storage medium. When the computer program is executed, it can include the processes of the embodiments of the above-mentioned methods. Among them, any reference to the memory, database or other medium used in the embodiments provided in the present application can include at least one of non-volatile and volatile memory. Non-volatile memory can include read-only memory (ROM), magnetic tape, floppy disk, flash memory, optical memory, high-density embedded non-volatile memory, resistive random access memory (ReRAM), magnetoresistive random access memory (MRAM), ferroelectric random access memory (FRAM), phase change memory (PCM), graphene memory, etc. Volatile memory can include random access memory (RAM) or external cache memory, etc. As an illustration and not limitation, RAM can be in various forms, such as static random access memory (SRAM) or dynamic random access memory (DRAM). The database involved in each embodiment provided in this application may include at least one of a relational database and a non-relational database. Non-relational databases may include distributed databases based on blockchains, etc., but are not limited to this. The processor involved in each embodiment provided in this application may be a general-purpose processor, a central processing unit, a graphics processor, a digital signal processor, a programmable logic device, a data processing logic device based on quantum computing, etc., but are not limited to this.

It should be noted that the user information (including but not limited to user device information, user personal information, etc.) and data (including but not limited to data used for analysis, stored data, displayed data, etc.) involved in this application are all information and data authorized by the user or fully authorized by all parties.

The technical features of the above embodiments may be arbitrarily combined. To make the description concise, not all possible combinations of the technical features in the above embodiments are described. However, as long as there is no contradiction in the combination of these technical features, they should be considered to be within the scope of this specification.

The above-described embodiments only express several implementation methods of the present application, and the descriptions thereof are relatively specific and detailed, but they cannot be understood as limiting the scope of the present application. It should be pointed out that, for a person of ordinary skill in the art, several variations and improvements can be made without departing from the concept of the present application, and these all belong to the protection scope of the present application. Therefore, the protection scope of the present application shall be subject to the attached claims.

Claims (9)

1. A data partitioning method in a micro-batch stream processing system, characterized in that the method comprises: Get data stream tuple; The data stream tuple is maintained based on a hash table and a balanced binary search tree; wherein the hash table stores the key of the data stream tuple, a first pointer pointing to a tuple list corresponding to the key, and a frequency count of the key; the frequency count of the key is also saved to the balanced binary search tree; each key in the hash table has a second pointer pointing to a corresponding frequency count node in the balanced binary search tree; Traversing the balanced binary search tree to generate an ordered list; wherein the ordered list includes the key, the frequency count of the key and the tuple list corresponding to the key; Based on a preset partitioning condition, the data stream tuples in the ordered list are partitioned into batches; wherein each partition is a data block, and each data block stores information on whether a key is split; all data stream tuples sharing the same key value are modeled as a single item, and the preset partitioning condition includes: limiting the number of split times of a single item, minimizing the number of different single items in a data block, and maintaining the capacity of each data block equal; wherein the batch partitioning problem is defined as a balanced bin packing problem for splittable items, given a set of N different items: k ₁ , k ₂ …, k _N , each item is of size _Sn , where 1≤n≤N, given B＝{b ₁ ,b ₂ ,…,b _B } boxes, each box has a capacity of C, the balanced bin packing problem for splittable items is to allocate each item to different boxes while satisfying the following conditions: (1) for any b _j , j∈[1,B], the number of tuples in the box is equal to the capacity C of the box; (2) for any b _j , j∈[1,B], the number of different items in the box is greater than or equal to N/B; (3) For any item, it is required to be split as few times as possible; when packing, split the data stream tuple corresponding to the key whose frequency count _i is greater than the ratio of the data block size to the data block cardinality into two items, one of which has a data stream tuple size equal to the ratio of the data block size to the data block cardinality, put it into the data block, and the other item is put into a new list; assign the remaining keys in the sorted list to the data blocks in a serpentine arrangement, and then assign the keys in the new list to the data blocks according to the best fit algorithm; The Map task uses the information of whether the key in the data block is split to allocate key clusters to the buckets of the Reduce stage for processing based on the worst fit algorithm; wherein the output of the Map stage is a cluster composed of key values, each key cluster has all data values of the same key; the capacity of the bucket is determined according to the ratio of the number of key clusters to the number of buckets; wherein the Map task uses the information that the key is split to allocate key clusters to the buckets of the Reduce stage for processing; wherein the output of the Map stage is a cluster composed of key-value pairs, each key cluster has all data values of the same key, and the key cluster C _k is represented as C _k ={(k,v _i )|v _i ∈k}, vi is the data value corresponding to key k; when the K key clusters output by a given Map stage need to be allocated to r Reduce buckets, the output of the Map task is I ={C _k |k∈K}, and the capacity of the bucket is set to The partitioning problem in the processing stage is simplified to the bin packing problem, where key clusters are regarded as items and Reducebuckets are bins. Given a set with M items and A bins a ₁ , a ₂ , … , a _A , each with capacity C _i , the variable-capacity balanced bin packing problem is to distribute items to different bins a ₁ , a ₂ , … , a _A under the following conditions: (1) for any a _j , j∈[1,A], the number of tuples in the bin is less than the capacity C _j of the bin; (2) for any a _j , j∈[1,A], the number of different items in the bin is greater than or equal to M. 2. The method according to claim 1, characterized in that the method further comprises: Record the processing time and interval of each batch; Obtaining the ratio of each batch processing time to the batch interval; According to a preset ratio threshold, obtaining a continuous batch count whose ratio satisfies the preset ratio threshold; According to the continuous batch count, the Map task and/or the Reduce task are adjusted. 3. The method according to claim 2, characterized in that the preset ratio threshold comprises a first ratio threshold; and obtaining the continuous batch count whose ratio satisfies the preset ratio threshold according to the preset ratio threshold comprises: A count of consecutive batches in which the ratio is greater than the first ratio threshold is obtained. 4. The method according to claim 3, characterized in that adjusting the Map task and/or Reduce task according to the continuous batch count comprises: When the first continuous batch count reaches a preset count threshold, Map tasks are added when the data rate increases, and Reduce tasks are added when the data distribution increases; wherein the first continuous batch count is a continuous batch count whose ratio is greater than the first ratio threshold. 5. The method according to claim 2, characterized in that the preset ratio threshold includes a second ratio threshold; and obtaining the continuous batch count whose ratio satisfies the preset ratio threshold according to the preset ratio threshold comprises: A count of consecutive batches in which the ratio is less than the second ratio threshold is obtained. 6. The method according to claim 5, characterized in that adjusting the Map task and/or Reduce task according to the continuous batch count comprises: When the second continuous batch count reaches a preset count threshold, the Map task is reduced when the data rate decreases, and the Reduce task is reduced when the data distribution decreases; wherein the second continuous batch count is a continuous batch count whose ratio is less than the second ratio threshold. 7. A dynamic data partitioning device in a micro-batch stream processing system, characterized in that the device comprises: The acquisition module is used to obtain the data stream tuple; A maintenance module, for maintaining the data stream tuple based on a hash table and a balanced binary search tree; wherein the hash table stores the key of the data stream tuple, a first pointer pointing to a tuple list corresponding to the key, and a frequency count of the key; the frequency count of the key is also saved to the balanced binary search tree; each key in the hash table has a second pointer pointing to a corresponding frequency count node in the balanced binary search tree; A generating module, configured to traverse the balanced binary search tree and generate an ordered list; wherein the ordered list comprises the key, the frequency count of the key and a tuple list corresponding to the key; A partitioning module is used to partition the data stream tuples in the ordered list into batches based on a preset partitioning condition; wherein each partition is a data block, and each data block stores information on whether a key is split; all data stream tuples sharing the same key value are modeled as a single item, and the preset partitioning condition includes: limiting the number of split times of a single item, minimizing the number of different single items in a data block, and maintaining the capacity of each data block equal; wherein the batch partitioning problem is defined as a balanced packing problem of splittable items, given a set of N different items: k ₁ , k ₂ , …, k _N , each item has a size of _Sn , where 1≤n≤N, given B＝{b ₁ ,b ₂ ,…,b _B } boxes, each box has a capacity of C, the balanced packing problem of splittable items is to distribute each item to different boxes while satisfying the following conditions: (1) for any b _j , j∈[1,B], the number of tuples in the box is equal to the capacity C of the box; (2) for any b _j , j∈[1,B], the number of different items in the box is greater than or equal to N/B; (3) For any item, it is required to be split as few times as possible; when packing, split the data stream tuple corresponding to the key whose frequency count _i is greater than the ratio of the data block size to the data block cardinality into two items, one of which has a data stream tuple size equal to the ratio of the data block size to the data block cardinality, put it into the data block, and the other item is put into a new list; assign the remaining keys in the sorted list to the data blocks in a serpentine arrangement, and then assign the keys in the new list to the data blocks according to the best fit algorithm; An allocation processing module is used to allocate key clusters to the buckets of the Reduce stage for processing by using the information of whether the key in the data block is split based on the worst fit algorithm through the Map task; wherein the output of the Map stage is a cluster composed of key values, each key cluster has all data values of the same key; the capacity of the bucket is determined according to the ratio of the number of key clusters to the number of buckets; wherein the Map task allocates key clusters to the buckets of the Reduce stage for processing by using the information that the key is split; wherein the output of the Map stage is a cluster composed of key-value pairs, each key cluster has all data values of the same key, and the key cluster C _k is represented as C _k ={(k,v _i )|v _i ∈k}, vi is the data value corresponding to key k; when K key clusters output by a given Map stage need to be allocated to r Reduce buckets, the output of the Map task is I ={C _k |k∈K}, and the capacity of the bucket is set to The partitioning problem in the processing stage is simplified to the bin packing problem, where key clusters are regarded as items and Reduce buckets are bins. Given a set with M items and A bins a ₁ , a ₂ , … , a _A , each with a capacity C _i , the variable-capacity balanced bin packing problem is to distribute items to different bins a ₁ , a ₂ , … , a _A under the following conditions: (1) for any a _j , j∈[1,A], the number of tuples in the bin is less than the capacity of the bin C _j ; (2) for any a _j , j∈[1,A], the number of different items in the bin is greater than or equal to M. 8. A computer device, comprising a memory and a processor, wherein the memory stores a computer program, wherein the processor implements the steps of the method according to any one of claims 1 to 6 when executing the computer program. 9. A computer-readable storage medium having a computer program stored thereon, wherein when the computer program is executed by a processor, the steps of the method according to any one of claims 1 to 6 are implemented.