TECHNICAL FIELD

This application relates to the field of information technologies, and in particular, to a data processing method and a system.

BACKGROUND

In the fields of finance, e-commerce, public security, and the like, there are scenarios such as fraud surveillance, accurate marketing recommendation, and crime and disaster forecast. In these scenarios, a user requires real-time analysis and processing on data that enters a database. For example, in real-time risk control of financial transactions, when each transaction is made, a risk control computing platform needs to calculate an aggregate value of all transactions within a 56-day time window in real time while meeting a delay requirement of a millisecond level, including a total transaction volume, a sum of squares of the transactions, a total quantity of the transactions, and the like. The aggregate value is used for further complex analysis such as rule inference and model scoring. When data is generated, value of the data decreases with time. Therefore, the data needs to be processed immediately when the data appears, instead of being cached for batch processing, and real-timeness of data processing by the database is particularly important.

In addition, the data in the foregoing scenarios features a relatively large scale and continuous arrival, and the features of the data pose a challenge to real-timeness of data processing. To process a large scale of continuous data in real time, and provide a user with an aggregate value of data in a database in real time, the data in the database needs to be aggregated before the database receives a query request. How to aggregate the data in the database to ensure real-timeness of a query request for the aggregate value and accuracy of a query range is a problem to be resolved urgently in a database system.

SUMMARY

According to a first aspect, an embodiment of this application provides a data processing method in a database. The method includes: obtaining at least one piece of data, where each piece of data includes at least one attribute; separately performing aggregation calculation, based on M aggregation intervals, on an attribute value of a first target attribute of the at least one piece of data to obtain T data clusters, where the M aggregation intervals are different from each other, each data cluster includes an aggregate value, the aggregate value of each data cluster is obtained by performing aggregation calculation on the attribute value of the first target attribute based on an aggregation interval corresponding to the data cluster, M is an integer greater than or equal to 2, and T≥M; and storing the T data clusters.

The plurality of aggregation intervals are selected, and the at least one piece of data in the database is aggregated based on the plurality of aggregation intervals, so that the data clusters stored in the database have data ranges with different attribute value spans. Therefore, when an aggregate value is queried based on a data cluster stored in the database, real-timeness of aggregate value query is improved while accuracy of a query range is ensured.

With reference to the first aspect, in a first possible implementation of the first aspect, the method further includes: presetting M aggregation layers, where each aggregation layer corresponds to one aggregation interval, the T data clusters belong to the M aggregation layers, any two adjacent aggregation intervals of the M aggregation intervals have a parent-child relationship, and a parent aggregation interval is an integer multiple of a child aggregation interval.

There is a plurality of methods for selecting the M aggregation intervals. Accuracy of the query range that can be queried depends on a minimum aggregation interval, and a parent aggregation interval of the minimum aggregation interval is used to further reduce time required for an aggregation operation, and improve the real-timeness of the aggregate value query.

With reference to the first aspect or the first possible implementation of the first aspect, in a second possible implementation of the first aspect, the obtaining at least one piece of data includes: receiving an Nthpiece of data in a data stream, where N is an integer greater than or equal to 1; and the separately performing aggregation calculation, based on M aggregation intervals, on an attribute value of a first target attribute of the at least one piece of data to obtain T data clusters includes: determining, based on an attribute value of a first target attribute of the Nthpiece of data, M data clusters respectively corresponding to the Nthpiece of data in the M aggregation layers; and if a data cluster corresponding to the Nthpiece of data in an mthaggregation layer has been generated in the database, refreshing an aggregate value of the generated data cluster based on the attribute value of the first target attribute of the Nthpiece of data, where m is a variable with a value range of 1≤m≤M; or if a data cluster corresponding to the Nthpiece of data in an mthaggregation layer has not been generated in the database, performing aggregation calculation on the attribute value of the first target attribute of the Nthpiece of data to generate a data cluster.

An aggregation operation is performed when each piece of data is received. Every time one piece of data is received, only M aggregation operations need to be performed, reducing an aggregation operation calculation amount, saving computing resources of a database system, and further improving real-timeness of query.

With reference to the first aspect or the first possible implementation of the first aspect, in a third possible implementation of the first aspect, the obtaining at least one piece of data includes: obtaining at least one piece of data stored in the database; and the separately performing aggregation calculation, based on M aggregation intervals, on an attribute value of a first target attribute of the at least one piece of data to obtain T data clusters includes: performing aggregation calculation on the attribute value of the first target attribute of the at least one piece of data based on an mthaggregation interval to obtain an mthaggregation layer, where the mthaggregation layer includes at least one data cluster, and m is a variable with a value range of 1≤m≤M.

With reference to any one of the first to the third possible implementations of the first aspect, in a fourth possible implementation of the first aspect, the method further includes: receiving a query request, where the query request includes a query range; obtaining, through query, at least two data clusters from the M aggregation layers based on the query range, where the at least two data clusters obtained through query belong to different aggregation layers, a data range of each data cluster obtained through query intersects with the query range, and a union set of the data ranges of all the data clusters obtained through query is the same as the query range; and performing aggregation calculation on aggregate values included in the at least two data clusters obtained through query, to obtain a query result.

Querying matched data clusters in different aggregation layers further reduces the time required for the aggregation operation and improves the real-timeness of the aggregate value query.

With reference to the fourth possible implementation of the first aspect, in a fifth possible implementation of the first aspect, the obtaining, through query, at least two data clusters from the M aggregation layers based on the query range specifically includes: obtaining, based on the query range and by performing matching layer by layer in descending order starting from a first aggregation layer, a data cluster that matches the query range in each aggregation layer, where the first aggregation layer is an aggregation layer with a largest aggregation interval.

With reference to any one of the first to the fifth possible implementations of the first aspect, in a sixth possible implementation of the first aspect, each aggregation layer further includes an aggregation algorithm of each aggregation layer.

With reference to any one of the first to the sixth possible implementations of the first aspect, in a seventh possible implementation of the first aspect, each aggregation layer further includes a data cluster list, and the data cluster list of each aggregation layer is used to store a data cluster included in the aggregation layer; and after the separately performing aggregation calculation, based on M aggregation intervals, on an attribute value of a first target attribute of the at least one piece of data to obtain T data clusters, the method further includes: updating a data cluster list of an aggregation layer to which each data cluster belongs.

With reference to any one of the first to the seventh possible implementations of the first aspect, in an eighth possible implementation of the first aspect, the method further includes: determining an index sequence number of each data cluster, where the index sequence number of each data cluster is used to indicate an order of the current data cluster in an aggregation layer to which the current data cluster belongs; and separately storing the index sequence number of each data cluster.

With reference to any one of the first aspect or the first to the eighth possible implementations of the first aspect, in a ninth possible implementation of the first aspect, the aggregation interval includes an aggregation time interval or an aggregation count value interval.

With reference to any one of the first aspect or the first to the ninth possible implementations of the first aspect, in a tenth possible implementation of the first aspect, the method further includes: setting a data channel, where the data channel includes an aggregation start mark, a capacity of the data channel, and a list of aggregation layers included in the data channel.

According to a second aspect, an embodiment of this application further provides a data processing method in a database. The database is used to store a plurality of pieces of data, and each piece of data includes at least one attribute. A data channel is set in the database, and the data channel includes M aggregation layers. Each aggregation layer corresponds to one aggregation interval, and M aggregation intervals are different from each other. The method includes: refreshing data clusters in the M aggregation layers when an Nthpiece of data in a data stream enters the data channel, where N is an integer greater than or equal to 1; and storing the M data clusters. The refreshing includes: obtaining, based on an attribute value of a first target attribute of the Nthpiece of data, M data clusters respectively corresponding to the Nthpiece of data in the M aggregation layers; and if a data cluster corresponding to the Nthpiece of data in an mthaggregation layer has been generated in the database, refreshing an aggregate value of the generated data cluster based on the attribute value of the first target attribute of the Nthpiece of data; or if a data cluster corresponding to the Nthpiece of data in an mthaggregation layer has not been generated in the database, performing aggregation calculation based on the attribute value of the first target attribute of the Nthpiece of data to generate a data cluster, where m is a variable with a value range of 1≤m≤M.

The plurality of aggregation intervals are selected, and the at least one piece of data in the database is aggregated based on the plurality of aggregation intervals, so that the data clusters stored in the database have data ranges with different attribute value spans. Therefore, when an aggregate value is queried based on a data cluster stored in the database, real-timeness of aggregate value query is improved while accuracy of a query range is ensured. An aggregation operation is performed when each piece of data enters the data channel. Every time one piece of data is received, only M aggregation operations need to be performed, reducing an aggregation operation calculation amount, saving computing resources of a database system, and further improving real-timeness of query.

With reference to the second aspect, in a first possible implementation of the second aspect, the data channel further includes an aggregation start mark and capacity information, and the method further includes: determining a data range of the data channel based on the aggregation start mark and a capacity of the data channel, and performing a step of refreshing the data clusters in the M aggregation layers after determining that an attribute value of a second target attribute of the Nthpiece of data belongs to the data range of the data channel.

Actually, a service requirement usually focuses only on an aggregate value of data that belongs to a specific data range in the database. The aggregation start mark and the capacity information of the data channel limit an amount of data that enters the data channel, reducing consumption of computing and storage resources of the database system.

With reference to the first possible implementation of the second aspect, in a second possible implementation of the second aspect, the method further includes: when the capacity information of the data channel and/or the aggregation start mark of the data channel is refreshed, deleting an aged data cluster at the M aggregation layers based on refreshed capacity information of the data channel or a refreshed aggregation start mark of the data channel.

Further, in some scenarios, the data range of the data channel changes with time or an amount of data that enters the database. With a change of the data range of the data channel, an aged data cluster appears in the data channel. Deleting the aged data cluster may further reduce consumption of the computing and storage resources of the database system.

With reference to any one of the second aspect or the first or the second possible implementation of the second aspect, in a third possible implementation of the second aspect, each aggregation layer includes a list of data clusters included in the aggregation layer. After the performing aggregation calculation based on the attribute value of the first target attribute of the Nthpiece of data to generate a data cluster, the method further includes: updating a data cluster list of the mthaggregation layer.

With reference to any one of the second aspect or the first to the third possible implementations of the second aspect, in a fourth possible implementation of the second aspect, the method further includes: determining an index sequence number of each of the M data clusters, where the index sequence number of each data cluster is used to indicate an order of the current data cluster in an aggregation layer to which the current data cluster belongs; and separately storing the index sequence number of each data cluster.

With reference to any one of the second aspect or the first to the fourth possible implementations of the second aspect, in a fifth possible implementation of the second aspect, the database stores T data clusters, where the T data clusters are obtained by separately performing aggregation calculation on an attribute value of a first target attribute of the at least one piece of data based on the M aggregation intervals, the T data clusters belong to the M aggregation layers, and T≥M. The method further includes: receiving a query request, where the query request includes a query range; obtaining, through query, at least two of the T data clusters from the M aggregation layers based on the query range, where the at least two data clusters obtained through query belong to different aggregation layers, a data range of each data cluster obtained through query intersects with the query range, and a union set of the data ranges of all the data clusters obtained through query is the same as the query range; and performing aggregation calculation on aggregate values included in the at least two data clusters obtained through query, to obtain a query result.

Querying matched data clusters in different aggregation layers further reduces time required for the aggregation operation and improves the real-timeness of the aggregate value query.

With reference to the fifth possible implementation of the second aspect, in a sixth possible implementation of the second aspect, the obtaining, through query, at least two of the T data clusters from the M aggregation layers based on the query range specifically includes: obtaining, based on the query range and by performing matching layer by layer in descending order starting from a first aggregation layer, a data cluster that matches the query range in each aggregation layer, where the first aggregation layer is an aggregation layer with a largest aggregation interval.

With reference to any one of the second aspect or the first to the sixth possible implementations of the second aspect, in a seventh possible implementation of the second aspect, the aggregation interval includes an aggregation time interval or an aggregation count value interval.

According to a third aspect, an embodiment of this application provides a data processing device, where the device includes an obtaining unit, an aggregation unit, and a storage unit. The obtaining unit is configured to obtain at least one piece of data, where each piece of data includes at least one attribute. The obtaining unit is configured to separately perform aggregation calculation, based on M aggregation intervals, on an attribute value of a first target attribute of the at least one piece of data to obtain T data clusters, where the M aggregation intervals are different from each other, each data cluster includes an aggregate value, the aggregate value of each data cluster is obtained by performing aggregation calculation on the attribute value of the first target attribute based on an aggregation interval corresponding to the data cluster, M is an integer greater than or equal to 2, and T≥M. The storage unit is configured to store the T data clusters.

The plurality of aggregation intervals are selected, and the at least one piece of data in a database is aggregated based on the plurality of aggregation intervals, so that the data clusters stored in the database have data ranges with different attribute value spans. Therefore, when an aggregate value is queried based on a data cluster stored in the database, real-timeness of aggregate value query is improved while accuracy of a query range is ensured.

With reference to the third aspect, in a first possible implementation of the third aspect, the obtaining unit is further configured to preset M aggregation layers, where each aggregation layer corresponds to one aggregation interval, the T data clusters belong to the M aggregation layers, any two adjacent aggregation intervals of the M aggregation intervals have a parent-child relationship, and a parent aggregation interval is an integer multiple of a child aggregation interval.

There is a plurality of methods for selecting the M aggregation intervals. Accuracy of the query range that can be queried depends on a minimum aggregation interval, and a parent aggregation interval of the minimum aggregation interval is used to further reduce time required for an aggregation operation, and improve the real-timeness of the aggregate value query.

With reference to the third aspect or the first possible implementation of the third aspect, in a second possible implementation of the third aspect, that the obtaining unit is configured to obtain at least one piece of data includes: receiving an Nthpiece of data in a data stream, where N is an integer greater than or equal to 1; and that the obtaining unit is configured to separately perform aggregation calculation, based on M aggregation intervals, on an attribute value of a first target attribute of the at least one piece of data to obtain T data clusters includes: determining, based on an attribute value of a first target attribute of the Nthpiece of data, M data clusters respectively corresponding to the Nthpiece of data in the M aggregation layers; and if a data cluster corresponding to the Nthpiece of data in an mthaggregation layer has been generated in the database, refreshing an aggregate value of the generated data cluster based on the attribute value of the first target attribute of the Nthpiece of data, where m is a variable with a value range of 1≤m≤M; or if a data cluster corresponding to the Nthpiece of data in an mthaggregation layer has not been generated in the database, performing aggregation calculation on the attribute value of the first target attribute of the Nthpiece of data to generate a data cluster.

An aggregation operation is performed when each piece of data is received. Every time one piece of data is received, only M aggregation operations need to be performed, reducing an aggregation operation calculation amount, saving computing resources of a database system, and further improving real-timeness of query.

With reference to the third aspect or the first possible implementation of the third aspect, in a third possible implementation of the third aspect, that the obtaining unit is configured to obtain at least one piece of data includes: obtaining at least one piece of data stored in the database; and that the obtaining unit is configured to separately perform aggregation calculation, based on M aggregation intervals, on an attribute value of a first target attribute of the at least one piece of data to obtain T data clusters includes: performing aggregation calculation on the attribute value of the first target attribute of the at least one piece of data based on an mthaggregation interval to obtain an mthaggregation layer, where the mthaggregation layer includes at least one data cluster, and m is a variable with a value range of 1≤m≤M.

With reference to any one of the first to the third possible implementations of the third aspect, in a fourth possible implementation of the third aspect, the device further includes a query unit. The query unit is configured to: receive a query request, where the query request includes a query range; obtain, through query, at least two data clusters from the M aggregation layers based on the query range, where the at least two data clusters obtained through query belong to different aggregation layers, a data range of each data cluster obtained through query intersects with the query range, and a union set of the data ranges of all the data clusters obtained through query is the same as the query range; and perform aggregation calculation on aggregate values included in the at least two data clusters obtained through query, to obtain a query result.

Querying matched data clusters in different aggregation layers further reduces the time required for the aggregation operation and improves the real-timeness of the aggregate value query.

With reference to the fourth possible implementation of the third aspect, in a fifth possible implementation of the third aspect, that the query unit is configured to obtain, through query, at least two data clusters from the M aggregation layers based on the query range specifically includes: obtaining, based on the query range and by performing matching layer by layer in descending order starting from a first aggregation layer, a data cluster that matches the query range in each aggregation layer, where the first aggregation layer is an aggregation layer with a largest aggregation interval.

With reference to any one of the first to the fifth possible implementations of the third aspect, in a sixth possible implementation of the third aspect, each aggregation layer further includes an aggregation algorithm of each aggregation layer.

With reference to any one of the first to the sixth possible implementations of the third aspect, in a seventh possible implementation of the third aspect, each aggregation layer further includes a data cluster list, and the data cluster list of each aggregation layer is used to store a data cluster included in the aggregation layer; and after the separately performing aggregation calculation, based on M aggregation intervals, on an attribute value of a first target attribute of the at least one piece of data to obtain T data clusters, the obtaining unit is further configured to update a data cluster list of an aggregation layer to which each data cluster belongs.

With reference to any one of the first to the seventh possible implementations of the third aspect, in an eighth possible implementation of the third aspect, the storage unit is further configured to: determine an index sequence number of each data cluster, where the index sequence number of each data cluster is used to indicate an order of the current data cluster in an aggregation layer to which the current data cluster belongs; and separately store the index sequence number of each data cluster.

With reference to any one of the third aspect or the first to the eighth possible implementations of the third aspect, in a ninth possible implementation of the third aspect, the aggregation interval includes an aggregation time interval or an aggregation count value interval.

With reference to any one of the third aspect or the first to the ninth possible implementations of the third aspect, in a tenth possible implementation of the third aspect, the obtaining unit is further configured to set a data channel, where the data channel includes an aggregation start mark, a capacity of the data channel, and a list of aggregation layers included in the data channel.

According to a fourth aspect, an embodiment of this application further provides a data processing device. The data processing device is configured to process at least one piece of data in a database, where each piece of data includes at least one attribute. A data channel is set in the database, and the data channel includes M aggregation layers. Each aggregation layer corresponds to one aggregation interval, and M aggregation intervals are different from each other. The device includes a data obtaining unit, an aggregation unit, and a storage unit. The obtaining unit is configured to obtain an Nthpiece of data in a data stream that enters the data channel. The obtaining unit is configured to refresh data clusters in the M aggregation layers, where N is an integer greater than or equal to 1. The refreshing includes: obtaining, based on an attribute value of a first target attribute of the Nthpiece of data, M data clusters respectively corresponding to the Nthpiece of data in the M aggregation layers; and if a data cluster corresponding to the Nthpiece of data in an mthaggregation layer has been generated in the database, refreshing an aggregate value of the generated data cluster based on the attribute value of the first target attribute of the Nthpiece of data; or if a data cluster corresponding to the Nthpiece of data in an mthaggregation layer has not been generated in the database, performing aggregation calculation based on the attribute value of the first target attribute of the Nthpiece of data to generate a data cluster, where m is a variable with a value range of 1≤m≤M. The storage unit is configured to store the M data clusters.

The plurality of aggregation intervals are selected, and the at least one piece of data in the database is aggregated based on the plurality of aggregation intervals, so that the data clusters stored in the database have data ranges with different attribute value spans. Therefore, when an aggregate value is queried based on a data cluster stored in the database, real-timeness of aggregate value query is improved while accuracy of a query range is ensured. An aggregation operation is performed when each piece of data enters the data channel. Every time one piece of data is received, only M aggregation operations need to be performed, reducing an aggregation operation calculation amount, saving computing resources of a database system, and further improving real-timeness of query.

With reference to the fourth aspect, in a first possible implementation of the fourth aspect, the data channel further includes an aggregation start mark and capacity information. The obtaining unit is further configured to: determine a data range of the data channel based on the aggregation start mark and a capacity of the data channel, and perform a step of refreshing the data clusters in the M aggregation layers after determining that an attribute value of a second target attribute of the Nthpiece of data belongs to the data range of the data channel.

Actually, a service requirement usually focuses only on an aggregate value of data that belongs to a specific data range in the database. The aggregation start mark and the capacity information of the data channel limit an amount of data that enters the data channel, reducing consumption of computing and storage resources of the database system.

With reference to the first possible implementation of the fourth aspect, in a second possible implementation of the fourth aspect, the storage unit is further configured to: refresh the capacity information of the data channel or the aggregation start mark of the data channel; and delete an aged data cluster at the M aggregation layers based on refreshed capacity information of the data channel or a refreshed aggregation start mark of the data channel.

Further, in some scenarios, the data range of the data channel changes with time or an amount of data that enters the database. With a change of the data range of the data channel, an aged data cluster appears in the data channel. Deleting the aged data cluster may further reduce consumption of the computing and storage resources of the database system.

With reference to any one of the fourth aspect or the first or the second possible implementation of the fourth aspect, in a third possible implementation of the fourth aspect, each aggregation layer includes a list of data clusters included in the aggregation layer. After the performing aggregation calculation based on the attribute value of the first target attribute of the Nthpiece of data to generate a data cluster, the storage unit is further configured to update a data cluster list of the mthaggregation layer.

With reference to any one of the fourth aspect or the first to the third possible implementations of the fourth aspect, in a fourth possible implementation of the fourth aspect, the storage unit is further configured to determine an index sequence number of each of the M data clusters, where the index sequence number of each data cluster is used to indicate an order of the current data cluster in an aggregation layer to which the current data cluster belongs; and the storage unit is further configured to separately store the index sequence number of each data cluster.

With reference to any one of the fourth aspect or the first to the fourth possible implementations of the fourth aspect, in a fifth possible implementation of the fourth aspect, the database stores T data clusters, where the T data clusters are obtained by separately performing aggregation calculation on an attribute value of a first target attribute of the at least one piece of data based on the M aggregation intervals, the T data clusters belong to the M aggregation layers, and T≥M. The device further includes a query unit, and the query unit is configured to: receive a query request, where the query request includes a query range; obtain, through query, at least two of the T data clusters from the M aggregation layers based on the query range, where the at least two data clusters obtained through query belong to different aggregation layers, a data range of each data cluster obtained through query intersects with the query range, and a union set of the data ranges of all the data clusters obtained through query is the same as the query range; and perform aggregation calculation on aggregate values included in the at least two data clusters obtained through query, to obtain a query result.

Querying matched data clusters in different aggregation layers further reduces time required for the aggregation operation and improves the real-timeness of the aggregate value query.

With reference to the fifth possible implementation of the fourth aspect, in a sixth possible implementation of the fourth aspect, that the query unit is configured to obtain, through query, at least two of the T data clusters from the M aggregation layers based on the query range specifically includes: obtaining, based on the query range and by performing matching layer by layer in descending order starting from a first aggregation layer, a data cluster that matches the query range in each aggregation layer, where the first aggregation layer is an aggregation layer with a largest aggregation interval.

With reference to any one of the fourth aspect or the first to the sixth possible implementations of the fourth aspect, in a seventh possible implementation of the fourth aspect, the aggregation interval includes an aggregation time interval or an aggregation count value interval.

According to a fifth aspect, an embodiment of this application provides a computing device. The computing device includes a processor and a memory, and the processor executes a program instruction in the memory to implement the various methods in the first aspect.

According to a sixth aspect, an embodiment of this application provides a computing device. The computing device includes a processor and a memory, and the processor executes a program instruction in the memory to implement the various methods in the second aspect.

According to a seventh aspect, an embodiment of this application provides a computer program product and a non-volatile computer readable storage medium, where the computer program product and the non-volatile computer readable storage medium include a computer instruction, and a computing device executes the computer instruction to implement the various methods in the first aspect of the embodiments of this application.

According to an eighth aspect, an embodiment of this application provides a computer program product and a non-volatile computer readable storage medium, where the computer program product and the non-volatile computer readable storage medium include a computer instruction, and a computing device executes the computer instruction to implement the various methods in the second aspect of the embodiments of this application.

