CN114942916A - Doris-based real-time data bin design method, device, equipment and storage medium - Google Patents

Abstract

The invention particularly discloses a Doris-based real-time digital storage design method, a Doris-based real-time digital storage design device, Doris-based real-time digital storage equipment and a Doris-based real-time digital storage medium. It includes: importing service real-time data into a Kakfa queue, putting the service real-time data into an adaptive topic tag, providing at least one consumption task for consuming the data in the topic tag, and setting adaptive route Load task monitoring; the route Load task monitors and reads the offset used for indicating the position information of the current data consumed in the topic label in real time; if the offset does not increase within the preset time, restarting the route Load task monitoring, and continuing to execute the consumption task from the offset; and importing the Stream Load data Stream after the consumption task is executed into Doris. The invention further provides a Doris importing technical means for real-time data, off-line data, fixed data and the like which need secondary business logic processing, and the method has great significance for increasing business support efficiency and reducing maintenance cost for warehouse construction.

Description

Doris-based real-time data warehouse design method, device, equipment and storage medium

technical field

The invention belongs to the technical field of data warehouses of logistics platforms, and in particular relates to a real-time data warehouse design method, device, equipment and storage medium of Doris.

Background technique

With the increasing growth of the express industry, the single-day ticket volume of logistics express data has reached 100 million. Faced with such a large amount of data, there are not only ticket data, but also intelligent hardware, such as vehicle cameras and other monitoring aspects. To process these data, an interactive analysis platform is required, so that business personnel can execute some Adhoc (multi-hop network) commands to query different data, and also support general SQL (Structured Query Language) syntax, reducing Learning costs, and need to query quickly.

Hanging or blocking often occurs during real-time data processing.

In addition, the daily data volume has reached more than 10 billion levels, which not only requires the performance of the analysis platform to be fast, but also requires detailed-level multi-tables of various ticket volumes to join (connect) multiple ticket-volume dimension tables of the billion level. . In addition, it is necessary to accurately deduplicate the amount of tickets with multiple tables, and ensure the accuracy of the data. At the same time, the platform also needs to support the export of various reports and the import of various data sources. The construction of warehouses on existing logistics platforms cannot meet the above requirements.

SUMMARY OF THE INVENTION

The present invention provides a Doris-based real-time data warehouse design method, device, equipment and storage medium, which solves the problem of hanging or blocking during real-time data import in the technology, and also provides a solution to support the import of more data sources.

This method includes:

Import the real-time business data into the Kakfa queue and put it into the adapted topic tag, provide at least one consumption task that consumes the data in the topic tag, and set the adapted Routine Load task monitoring;

Described Routine Load task monitoring reads in real time the offset that is used to indicate the offset of the location information where the current data is consumed in the topic label;

If the offset does not increase within the preset time, then restart the Routine Load task monitoring, and continue to execute the consumption task from the offset;

Import the Stream Load data stream after the execution of the consumption task into Doris.

Further, importing the Stream Load data after the execution of the consumption task into Doris further includes: creating a final output position of the Stream Load data stream on the real-time computing cluster Flink to the sink in the Doris: flink-connector- dorisdb; Stream Load data is directly imported into the Doris from Http through the cache and in batches.

This solution also includes real-time import of data processed by secondary business logic, including:

After the Stream Load data stream after the execution of the consumption task is imported into Doris, it also includes:

Find the real-time data that needs secondary business logic processing, first perform logical processing of the secondary business on the real-time data through Flink, then cache the logically processed data, and finally store the cached data by Stream Load import Doris. At least one real-time data that needs secondary business logic processing is first processed by Flink's secondary business logic and further includes: at least one secondary business logic of the real-time data is set to a task management, and the task management submits an adapted task job to the task management. JobManager. The JobManager is responsible for the scheduling and resource allocation of the job. The JobManager assigns the job to the corresponding TaskManager. After receiving the task job, the TaskManager starts a thread to execute it and reports to the JobManager. The task status and its own running status; when the task execution ends, the JobManager will receive a notification and send the data processed by the secondary business logic to the corresponding task manager.

In an example, at least one real-time data requiring secondary business logic processing is first processed by Flink's secondary business logic and further includes:

First consume data from the ods log topic, and then divide the behavior data in the data into three types through user behavior: page log, exposure log, and startup log;

By defining a Flink state variable as time, it is used to filter the data of the new user to determine whether it is a new user. If it is the first time, the corresponding state is empty, and the current time is updated to the state. Otherwise, the status is not empty, then the current user is modified to an old user;

Divide the data after filtering, write the startup log and the exposure data into the output stream, write the page data into the mainstream and obtain the corresponding value through json, and write the startup data in start. Describe the measurement output stream, remove all the startup logs from the page log, write the page log to the mainstream, and then write the exposure log to the exposure side output stream

The method also includes the import of offline business data, which includes:

In an offline business scenario, use Broker Load to import the offline business data into the Doris from HDFS;

Through SQL, use DolphinSchedule to configure scheduling tasks, complete the processing of the offline business data in batches, and then summarize them into a general table.

The method also includes relatively fixed data import, which further includes:

Relatively fixed data in actual business scenarios is directly connected to the data service hall, which is connected to Power BI tools and front-end reports.

Based on the same concept, the present invention also provides a Doris-based real-time data warehouse design device, including:

Routine Load task monitoring and setting module: used to import real-time business data into the Kakfa queue and put it into the adapted topic tag, then provide at least one consumption task that consumes the data in the topic tag, and set the adapted Routine Load task monitor;

Routine Load task monitoring and processing module: for the Routine Load task monitoring real-time read the offset for indicating the location information of the current data consumption in the topic label;

The Routine Load task monitoring startup module: for if the offset does not increase within a preset time, then restart the Routine Load task monitoring, and continue to execute the consumption task from the offset;

Import module: used to import the Stream Load data stream after the execution of the consumption task into Doris.

Based on the same concept, the present invention also provides a computer device, comprising:

a memory for storing the processing program;

A processor, when the processor executes the processing program, the above-mentioned Doris-based real-time data warehouse design method is implemented.

Based on the same concept, the present invention also provides a readable storage medium, on which a processing program is stored, and when the processing program is executed by a processor, the above-mentioned Doris-based real-time data warehouse design method is implemented.

After adopting the above-mentioned technical scheme, compared with the prior art, the beneficial part of the present invention is:

The present invention makes a self-defined Routine Load monitoring to read its offset in real time. If it is found that the offset no longer increases, it will manually execute the task of stopping the Routine Load through the monitoring system, and restart the Routine Load. The offset is used before, which can ensure that the data is not lost or repeated, and at the same time restores the Routine Load. The present invention improves the efficiency and security of real-time data import processing, and is not easy to block. There is no backlog of data, ensuring smooth operation during peak data access.

The present invention supports the export of various reports and the import of various data sources. It is of great significance for the construction of data warehouses to improve business support efficiency and reduce maintenance costs. For example, in complex business logic, real-time business data is first processed by Flink, and then cached by Redis or PIKA. After processing, the Stream Load method can be used to import Doris; in offline business scenarios, the business data uses Broker Load to transfer business data from HDFS imports Doris, and then uses DolphinScheduler to run batch processing of complex business data through SQL, and finally summarizes it into a summary table; in actual business scenarios, it is directly connected to the data service hall, and the data service hall is connected to Power BI tools and front-end reports. The amount of tickets with multiple tables is accurately deduplicated, and the accuracy of the data is guaranteed; it can fully support the export of multiple reports and the import of multiple data sources.

Description of drawings

The specific embodiments of the present invention will be described in further detail below in conjunction with the accompanying drawings, wherein:

Fig. 1 is the first example schematic diagram of the real-time data warehouse design method based on Doris of the present invention;

Fig. 2 is the principle block diagram of the real-time data warehouse design system based on Doris of the present invention;

Fig. 3 is the second example flow chart of the real-time data warehouse design method based on Doris of the present invention;

4 is a schematic diagram of an embodiment of a real-time data warehouse design device based on Doris according to the present invention;

FIG. 5 is a schematic diagram of an embodiment of a computer device of the present invention.

Detailed ways

The present invention will be further described in detail below with reference to the accompanying drawings and specific embodiments. The advantages and features of the present invention will become more apparent from the following description and claims. It should be noted that, the accompanying drawings are all in a very simplified form and all use imprecise ratios, and are only used to facilitate and clearly assist the purpose of explaining the embodiments of the present invention.

It should be noted that all directional indications (such as up, down, left, right, front, back, etc.) in the embodiments of the present invention are only used to explain the relationship between various components under a certain posture (as shown in the accompanying drawings). The relative positional relationship, the movement situation, etc., if the specific posture changes, the directional indication also changes accordingly.

Example 1

In the era of big data, the data processing capability is greatly enhanced, but there is still a big bottleneck in the last link, that is, the data application service link. It turns out that in the era of business databases, high concurrency and high flexibility are contradictory. How to make the front-line flexibly customize and analyze SQL (Structured Query Language), although drag and drop generates SQL that can’t bear to look at, but I hope the query engine can still maintain Strong performance indicators, whether it is concurrency or query time, can satisfy customers, which is quite a challenge. At present, there are a lot of query analysis, and they are still emerging. Engines emerge in an endless stream, each with its own advantages and disadvantages, such as ADB, Hologres, Presto, Kylin, Hbase (several database names), these products essentially use resources Changing time, or changing space for time, essentially means computing mechanism reconstruction, hardware acceleration (such as SSD disks, or solid-state drives), index acceleration (such as bitmaps), and space conversion (such as precomputing) to improve performance. The emerging and updated technology products also confirm from the side that there are still many problems with the query engine, the pain points are still unresolved, and there is still much to be done. Doris is a database system that has recently been adapted to real-time analysis scenarios and has indeed solved the pain points of query engines in some scenarios.

Doris consists of FrontEnd DorisDB (front-end node) and BackEnd DorisDB (back-end node) core components. The front-end node is responsible for managing metadata, managing client connections, query planning and scheduling; the back-end node is responsible for data storage, calculation execution, copy management, etc.; it also includes DorisManager (Doris management tool) and Broker (broker), The DorisManager management tool is responsible for providing visualization tools for cluster management, online query, fault query, monitoring and alarming; Broker is responsible for auxiliary functions such as data export and import with external storage (HDFS, Hadoop distributed file system or object storage); Doris can pass MySQL client direct access.

The core components and operating modes of Doris are as follows (hereinafter FrontEnd DorisDB referred to as FE, BackEndDorisDB referred to as BE):

FE is introduced first.

1) Manage metadata, execute SQL DDL commands, and use Catalog (record library), tables, partitions, tablet replicas and other information.

2) FE high-availability deployment, using the replication protocol to select the master and master-slave to synchronize metadata, all metadata modification operations are completed by the FE leader node (FE master node), and the FE follower node (FE tracking node) can perform read operations. Metadata reads and writes meet sequential consistency. The number of FE nodes is 2n+1, which can tolerate n node failures. When the FE leader fails, the existing follower node is re-elected to complete the failover.

3) FE's SQL layer (SQL layer) parses, analyzes, rewrites, semantically analyzes and optimizes relational algebra to the SQL submitted by the user, and produces a logic execution plan.

4) The Planner (SQL physical plan) of FE is responsible for transforming the logical plan into a physical plan that can be executed in a distributed manner and distributes it to a group of BEs.

5) FE supervises BE, manages the online and offline of BE, and maintains the number of tablets (replicas) according to the survival and health status of BE.

6) FE coordinates data import to ensure the consistency of data import.

Next, we introduce BE.

1) BE manages the tablet (copy), the tablet is a sub-table formed by partitioning and bucketing, and it is stored in a columnar format.

2) BE is guided by FE to create or delete child tables.

3) BE receives the physical execution plan distributed by FE and designates the BE coordinator node (BE coordination adjustment point), and under the scheduling of the BE coordinator, it cooperates with other BE workers (BE execution nodes) to complete the execution.

4) BE reads the local column storage engine to obtain data, and quickly filters the data through index and predicate sinking.

5) BE performs compact (compression) tasks in the background to reduce read amplification during query.

Taking the query as an example, through the organization, coordination and control of FE, the submitted SQL is parsed, analyzed, rewritten, optimized and planned to generate a distributed execution plan, which is then executed by several BEs, and a coordinator is selected among several BEs (collaborator), the collaborator coordinates n BEs for local calculation, and then returns to the collaborator, the collaborator returns the final result to the FE after summarizing, and finally the FE provides the final result to the end user.

This technical solution provides a first example schematic diagram of a Doris-based real-time data warehouse design method (see Figure 1), including importing various data into the Doris data warehouse in different ways:

S1: Import of real-time data: import real-time business data from Flume (mass log collection, aggregation and transmission system) or Canal (Alibaba's open source middleware) to Kafka (a high-throughput distributed publish-subscribe messaging system) , The real-time business data in Kafka (a distributed log system) undergoes multiple data processing, and is imported into Doris using SQL after data processing.

S2: Importing complex real-time data: Generally, real-time data needs to be processed by secondary logic, which is called complex real-time data business logic. The real-time data of this type of business is first processed by Flink (a distributed processing engine for streaming data and batch data, that is, a real-time computing cluster), and then cached by Redis or PIKA (a redis-like storage system open source by Qihoo), and the processing is completed. You can either import Doris with the Stream Load (a synchronous import method) method. Specifically, first find the real-time data that needs secondary business logic processing, first perform logical processing of the secondary business on the real-time data through Flink, then cache the logically processed data, and finally cache the cached data. The data is imported into Doris by Stream Load.

S13: Import of offline business data: The business data uses Broker Load (an asynchronous import method) to import business data from HDFS (Hadoop Distributed File System) into Doris, and then uses SQL to run batches using DolphinScheduler (batch plan). Complex business data is processed and finally aggregated into a summary table. From another point of view, if the data source is Hive/HDFS, Broker Load is recommended for import. If it is troublesome to import a lot of data tables, you can consider using Hive external direct connection query. The performance will be worse than Broker Load import, but it can avoid data relocation. If the amount of data in a single table is particularly large, or you need to do a global data dictionary to accurately deduplicate, you can consider Spark Load import.

S14: Import of fixed data (such as special forms, etc.): In actual business scenarios, directly connect to the data service hall, and the data service hall connects to Power BI tools and front-end reports. The amount of tickets in multiple tables is accurately deduplicated, and the accuracy of the data is ensured; the proposal of the present invention supports the export of various reports and the import of various data sources; there is no backlog of data, ensuring smooth operation during peak data access.

In the present invention, an import operation can be divided into several stages:

1) LOADING

This stage cleans and transforms the data first, and then sends the data to BE for processing. After all the data is imported, the process of waiting for validation is entered. At this time, the status of the import job is still LOADING.

2)FINISHED

After all the data involved in the import job takes effect, the job status becomes FINISHED, and the imported data after FINISHED can be queried. FINISHED is the final status of the import job.

3)CANCELLED

Before the import job status becomes FINISHED, the job may be canceled and enter the CANCELLED state at any time, such as manual cancellation by the user or an error in the import. CANCELLED is also a final state of an import job.

The applicable scenarios of the import operation of the present invention are as follows:

1) HDFS import

When the source data is stored in HDFS, and the data volume ranges from tens of GB to hundreds of GB, the Broker Load method can be used to import data to DorisDB. At this point, the deployed Broker process is required to have access to the HDFS data source. The job of importing data is executed asynchronously, and users can view the import result through the SHOW LOAD command.

The source data is stored in HDSF. When the data volume reaches the TB level, the Spark Load method can be used to import data to DorisDB. At this point, the deployed Spark process is required to have access to the HDFS data source. The job of importing data is executed asynchronously, and users can view the import result through the SHOW LOAD command.

For other external data sources, as long as the Broker or Spark process can read the corresponding data source, the Broker Load or Spark Load method can also be used to import data.

2) Local file import

The data is stored in local files, and the data volume is less than 10 GB. The Stream Load method can be used to quickly import the data into the DorisDB system. The import job is created using the HTTP protocol, the job is executed synchronously, and the user can judge whether the import is successful through the return value of the HTTP request.

3) Kafka import

The data comes from streaming data sources such as Kafka. When real-time data needs to be imported into the DorisDB system, the Routine Load method can be used. Users create routine import jobs through the MySQL protocol, and DorisDB continuously reads and imports data from Kafka.

4) Insert Into import

The Insert Into method can be used to write data to the DorisDB table during manual testing and temporary data processing. Among them, the INSERT INTO tbl SELECT...; statement reads data from the DorisDB table and imports it into another table; the INSERT INTO tbl VALUES(...); statement inserts a single piece of data into the specified table.

5) At the same time, there are other ways to import Json data (for some semi-structured data such as Json type, we can use stream load or routine load to import. Stream Load: For Json data stored in text files, we You can use stream load to import. Routine Load: For json format data in Kafka, you can use Routine load to import), flink-connector-dorisdb (the internal implementation is through cache and batch import by stream load), DataX-dorisdb- The writer (DorisWriter plug-in implements the function of writing data to the destination table of DorisDB. In the underlying implementation, DorisWriter imports data to DorisDB in csv or json format through Streamload. Internally, the data read by the reader is cached and then imported into DorisDB in batches. To improve write performance. The overall data flow is source->Reader->DataX channel->Writer->DorisDB.)

Data export (Export) is a function provided by DorisDB to export and store data to other media. This function can export the data of the table or partition specified by the user to the remote storage, such as HDFS/Alibaba Cloud OSS/AWS S3 (or object storage compatible with the S3 protocol), through the Broker process in text format.

After a user submits an export job, DorisDB will count all the Tablets involved in the job, then group these Tablets, and generate a special query plan for each group. The query plan will read the data on the included Tablet, and then write the data to the path specified by the remote storage through the Broker. The processing flow mainly includes: 1) The user submits an Export job to the FE.

2) FE's export scheduler will execute an export job in two stages: 3) PENDING: FE generates an ExportPendingTask, sends a snapshot command to BE, takes a snapshot of all involved Tablets, and generates multiple query plans. 4) EXPORTING: FE generates an ExportExportingTask and starts to execute query plans one by one.

Example 2

Please refer to FIG. 2 , which is a schematic system diagram of a Doris-based real-time data warehouse design example of the present invention. It includes data center 11, Doris data warehouse 12, Flink real-time computing cluster 14 (hereinafter referred to as Flink), Kafka13, HDFS15. The data warehouse as a whole builds an enterprise-level data warehouse with Doris12 as the core, (through the data collection system, a variety of data collection Means, including Mysql binlog analysis (Cannal), log collection can realize the collection of various heterogeneous data through Flume (Doris audit log), buried point interface, etc., and the collected data is unified through the message queue (Kafka13) to complete highly concurrent data Throughput, realize the decoupling of data warehouse and computing engine at the same time. Flink14 completes data ETL processing and real-time data statistics. Data center 11 completes the management and control of the above matters, including data quality management and data storage and data service life cycle data Management, which is to realize the data management of the whole life cycle from data collection to data ETL processing, data storage and data service, including metadata, data quality, data specification, data security, etc. HDFS15 is to realize offline data storage, used for Import offline data into Doris 11. The data center 11 can be implemented by many server clusters or servers, which will be referred to as servers for the convenience of description later.

There are many kinds of above-mentioned principle systems for realizing real-time data warehouse design, and the above is just an example to illustrate.

Please refer to Figure 3, a flow chart of a real-time data warehouse design method based on Doris. include:

S110: import the real-time business data into the Kakfa queue, and put it into the adapted topic label, provide at least one consumption task for consuming the data in the topic label, and set the adapted Routine Load task monitoring;

S120: Routine Load task monitoring real-time reads the offset used to indicate the location information of the current data consumption in the topic tag;

S130: If the offset does not increase within a preset time, restart the Routine Load task monitoring, and continue to execute the consumption task from the offset;

S140: Import the Stream Load data stream after the execution of the consumption task into Doris.

This process is for real-time business data that does not require special complex secondary processing to be imported into Doris in real time. Multiple topic tags are set in the Kakfa queue (as shown in FIG. 1 , including the first topic tag 131 , the second topic tag 132 , and the Nth topic tag 133 ). The real-time business data is imported into the Kakfa queue on demand or by category, and placed in the appropriate topic tag. If the Stream Load data stream of a topic that is directly consumed is directly imported into Doris, it will be found that the data will be blocked and often hang up. For this reason, our technicians have analyzed and learned that due to the large amount of imported data and the large number of transactions, even if the problem of backlog data is found. To this end, our company sets up RoutineLoad task monitoring on the server. The consumption task that provides the data in the consumption topic tag is generally adapted to a Routine Load task monitoring. In the present invention, in the first topic tag 131, the second topic tag 132, ... the Nth topic tag 133, there may be multiple consumption tasks for consuming data in a topic tag, for example, the first topic tag 131 may have multiple consumption tasks tasks, and new consuming tasks are created in real time as needed for processing or statistics. In the present invention, the corresponding Routine Load task monitoring is created for the message task, and when the offset in the Routine Load task monitoring does not change within a preset time (such as a few seconds), the Routine Load is restarted Task monitoring: Continue to execute the consumption task according to the offset, that is, restart the execution of the corresponding Stream Load data stream in the consumer task and import it into Doris. In this example, a Routine Load task monitoring group can be established, and each topic label 131 is set with a Routine Load task monitoring group. When there are Load task monitoring groups within a preset time, if there are multiple groups of offsets that do not change, you can Start various Routine Load task monitoring at the same time. When this kind of state change occurs many times within a preset time, a notification that needs to be processed manually can be issued. This setting further improves the processing efficiency and enhances its security. In this example, a custom Routine Load monitoring is done to read its offset in real time. If it is found that the offset is no longer growing, it will manually execute the task of stopping Routine Load through the monitoring system, and restart Routine Load. The offset is used before, which can ensure that the data is not lost or repeated, and the Routine Load can be restored at the same time.

In addition, in this example, the most commonly used Stream Load can be customized through the sink of Flink. This operation is relatively simple, and the data is directly imported into Doris using http. That is, importing the Stream Load data after the execution of the consumption task into Doris further includes:

Create a final output position of the Stream Load data stream on the Flink real-time computing cluster to the sink in the Doris: flink-connector-dorisdb;

Stream Load data is directly imported into the Doris through Http through the cache and in batches.

Specifically, first download the jar package of the connector and put it in the lib/ directory of flink, then enter the cli and execute the following command to create a sink to dorisdb. Create a flink-connector-dorisdb: the final output location of the Stream Load data stream to the sink in Doris. Adding an exact-once data sink requires the two phase commit mechanism of the external system. Since DorisDB does not have this mechanism, the applicant needs to rely on flink's checkpoint-interval to save the cached data group and its label at each checkpoint, and block the cached data in all states from flushing when the subsequent data is incoming, so as to achieve Precise once. However, if DorisDB hangs up, the user's flink sink stream operator will be blocked for a long time, and will cause flink's monitoring alarm or forced kill. The present invention can use the csv format for import by default. If the import still exists, the present invention can also increase the memory of the flink task to improve the work efficiency.

Example three

The above example 2 is mainly for importing real-time data for business. In this example 3, the real-time data can be added to Doris after secondary business logic processing. Before real-time import, there are still a lot of data or tasks that need various secondary processing to import, in this case, the present invention adopts the following scheme to realize:

At least one real-time data that needs secondary business logic processing is first processed by Flink's secondary business logic, then cached, and then the Stream Load data after processing is imported into Doris. Specifically, it also includes: finding the real-time data that needs secondary business logic processing, first performing the logic processing of the secondary business on the real-time data through Flink, then caching the logically processed data, and finally caching the data. The data is imported into Doris by Stream Load. For example, if there are a large number of duplicate data records in the data, the secondary business logic processing can include checking and de-duplication processing of these data, and then cache the checked and de-duplicated data first, and then import it through Stream Load. Doris.

At least one real-time data requiring secondary business logic processing is first processed by Flink's secondary business logic and further includes:

The secondary business logic of at least one of the real-time data is set to a task management, and the task management submits an adapted task job to the JobManager, and the JobManager is responsible for the scheduling and resource allocation of the job,

Described JobManager assigns described job to corresponding TaskManager, after TaskManager receives described task job, starts thread to carry out, and reports described task state and self-running state to JobManager;

After described task execution ends, described JobManager will receive notification, and send the data after secondary business logic processing to corresponding task management.

For example, at least one real-time data that needs secondary business logic processing is first processed by Flink's secondary business logic and further includes:

First consume data from the ods log topic, and then divide the behavior data in the data into three types through user behavior: page log, exposure log, and startup log;

By defining a state variable of flink as time, it is used to filter the data of new users: to determine whether it is a new user, if it is the first time, its state is empty, and the current time is updated into the state, otherwise the state If it is not empty, the current user is changed to an old user;

Divide the filtered data, write the startup log and the exposure data into the output stream on the side, write the page data into the mainstream and obtain the corresponding value through json, and write the startup test output stream in start , remove all the startup logs as page logs, write the page logs into the main stream, and then determine whether they are exposure logs, and then write them into the exposure side output stream.

Example 4

Compared with the second example or the third example, the fourth example includes the processing of the introduction scheme of the offline service. It further includes: in the offline business scenario, using Broker Load to import the offline business data from HDFS into the Doris; using SQL mode, using DolphinScheduler to configure scheduling tasks, batch processing the offline business data, and then summarize it into in a summary table.

Configure scheduling tasks in DolphinScheduler, that is, which offline services need to import the configured scheduling tasks, the corresponding data can be imported to Doris through HDFS, and can be aggregated into a summary table as required. Also, query the imported results with the following command

show load where label="ts_202111221719";

CANCELLED: Import failed, you can view the reason for the error reported by the http link

LOADING: Loading normally.

FINISHED: Import complete

The above commands are used to track and feedback each scheduling task, and the imported results can be obtained and carried out in time.

Example 5

This example also includes: the relatively fixed data in the actual business scenario is directly connected to the data service hall, and the data service hall is connected to the Power BI tool and the front-end report. The purpose of this step is to directly output some relatively fixed data to the front-end report. For Adhoc data, BI tools are provided for business personnel to query. PowerBI is an extraordinary product from Microsoft. Power BI is an advanced function based on Excel. It extends and develops a variety of function points as a collection, and finally creates this BI tool. Power BI can directly import and analyze data in conventional file formats, such as the well-known Excel, TXT, etc., and can also directly connect to traditional databases and multidimensional databases. It cannot be said that Power BI, as the handwriting of a major manufacturer, is very stable and smooth to use. In addition, PowerBI provides some visual interface operation options, combined with M language and DAX functions, data processing can be performed.

Embodiment 6

As shown in Figure 4, based on the same concept, the present invention also a real-time data warehouse design device based on Doris, comprising:

Routine Load task monitoring and setting module 401: used to import real-time business data into the Kakfa queue and put it into the adapted topic tag, then provide at least one consumption task for consuming the data in the topic tag, and set the adapted Routine Load task monitoring;

Routine Load task monitoring and processing module 402: used for the Routine Load task monitoring to read in real time the offset used to indicate the location information of the current data consumption in the topic tag;

The Routine Load task monitoring startup module 403: for if the offset does not increase within a preset time, then restart the Routine Load task monitoring, and continue to execute the consumption task from the offset;

Import module 404: used to import the Stream Load data stream after the execution of the consumption task into Doris.

The real-time data warehouse design device can be integrated into the above-mentioned data warehouse, or can be a separate server.

Embodiment 7

As shown in FIG. 5, based on the same concept, the present invention also provides a computer device 600. The computer device 600 may vary greatly due to different configurations or performances, and may include one or more processors 610 (central processing units, CPU) 710 (eg, one or more processors) and memory 620, one or more storage media 630 (eg, one or more mass storage devices) storing application programs 633 or data 632. Among them, the memory 620 and the storage medium 630 may be short-term storage or persistent storage. The program stored in the storage medium 630 may include one or more modules (not shown), and each module may include a series of instructions to operate on the computer device 600. Furthermore, the processor 610 may be configured to communicate with the storage medium 630 to execute a series of instruction operations in the storage medium 630 on the computer device 600 .

Computer device 600 may also include one or more power supplies 640, one or more wired or wireless network interfaces 650, one or more input and output interfaces 660, and/or, one or more operating systems 631, such as Windows Server, Mac OS X, Unix, Linux, FreeBSD and more. Those skilled in the art can understand that the computer device structure shown in FIG. 5 does not constitute a limitation to the computer device, and may include more or less components or combine some components, or arrange different components than those shown.

When the computer-readable instruction is executed by the processor, by importing the business real-time data from Flume or Canal into Kafka, the business real-time data in Kafka is processed multiple times, and the data is processed and imported into Doris with SQL; In the business logic, the real-time business data is first processed by Flink, and then cached by Redis or PIKA. After the processing is completed, the Stream Load method can be used to import it into Doris; in the offline business scenario, the business data is imported from HDFS to Doris by BrokerLoad, and then Through SQL, use DolphinScheduler to run batch processing of complex business data, and finally summarize it into a summary table; in actual business scenarios, it is directly connected to the data service hall, which is connected to Power BI tools and front-end reports. Accurately deduplicate the amount of tickets with multiple tables, and ensure the accuracy of the data; support the export of various reports and the import of various data sources; no data backlog, ensuring smooth operation during peak data volume access.

In one embodiment, a readable storage medium is provided. When the computer-readable instructions are executed by one or more processors, the one or more processors execute the above-mentioned method for designing a real-time data warehouse based on Doris , the specific steps are not repeated here.

Those skilled in the art can clearly understand that, for the convenience and brevity of description, the specific working process of the above-described system, device and unit can refer to the corresponding process in the foregoing method embodiments, which will not be repeated here.

The integrated unit, if implemented in the form of a software functional unit and sold or used as a stand-alone product, may be stored in a computer-readable storage medium. Based on this understanding, the technical solution of the present invention is essentially or the part that contributes to the prior art, or all or part of the technical solution can be embodied in the form of a software product, and the computer software product is stored in a storage medium , including several instructions for causing a computer device (which may be a personal computer, a server, or a network device, etc.) to execute all or part of the steps of the methods described in the various embodiments of the present invention. The aforementioned storage medium includes: a U disk, a removable hard disk, a read-only memory (ROM), a random access memory (RAM), a magnetic disk or an optical disk and other media that can store program codes.

As mentioned above, the above embodiments are only used to illustrate the technical solutions of the present invention, but not to limit them; although the present invention has been described in detail with reference to the foregoing embodiments, those of ordinary skill in the art should understand: The technical solutions described in the embodiments are modified, or some technical features thereof are equivalently replaced; and these modifications or replacements do not make the essence of the corresponding technical solutions depart from the spirit and scope of the technical solutions of the embodiments of the present invention.

Claims (10)

1. A real-time data bin design method based on Doris is characterized by comprising the following steps:

importing service real-time data into a Kakfa queue, putting the service real-time data into an adaptive topic tag, providing at least one consumption task for consuming the data in the topic tag, and setting adaptive route Load task monitoring;

the route Load task monitoring real-time reading is used for indicating the offset of the position information of the consumption current data in the topic tag;

if the offset does not increase within the preset time, restarting the route Load task monitoring, and continuing to execute the consumption task from the offset;

and importing the Stream Load data Stream after the consumption task is executed into Doris.

2. The Doris-based real-time data bin design method of claim 1, wherein the importing the Stream Load data after the consuming task is executed into Doris further comprises:

creating a final output position of the Stream Load data Stream on a real-time computing cluster Flink to a sink in the Doris: (ii) flink-connector-dorisdb;

the Stream Load data is directly imported into the Doris by Http in batches through caching.

3. The Doris-based real-time data warehouse design method of claim 1, wherein the importing the Stream Load data Stream after the consuming task is executed into Doris further comprises:

and finding real-time data needing secondary service logic processing, firstly carrying out the logic processing of the secondary service on the real-time data through a flash, then caching the data after the logic processing, and finally leading the cached data into Doris by Stream Load.

4. The Doris-based real-time data bin design method according to claim 3, wherein the first logic processing of the secondary service on the real-time data through Flink further comprises:

setting a secondary business logic of the real-time data into a task management, submitting the adapted task job to a JobManager, wherein the JobManager is responsible for scheduling and resource allocation of the job,

the JobManager distributes the operation to a corresponding TaskManager, the TaskManager starts a thread to execute after receiving the task operation, and reports the task state and the self running state to the JobManager;

and after the task execution is finished, the JobManager receives the notification and sends the data after the secondary service logic processing to the corresponding task management.

5. The Doris-based real-time data bin design method according to claim 3 or 4, wherein the first performing the logic processing of the secondary service on the real-time data through Flink, wherein the logic processing of the secondary service further comprises:

data is consumed from the ods log topic, and then behavior data in the data is divided into 3 types through user behaviors: page log, exposure log, start log;

defining a state variable of the Flink as time, wherein the state variable is used for filtering the data of a new user and judging whether the data is the new user, if the data is the new user, the corresponding state is empty, the current time is updated to the state, and if the data is not the new user, the current user is modified to an old user;

the data after being filtered is subjected to shunting processing, the starting log and the exposure data are written into an output stream at an exposure side, the page data are written into the main stream and corresponding values are obtained through json, the output stream is written into a start to start, the starting log is removed and is completely the page log, the page log is written into the main stream, and then the exposure log is written into the output stream at the exposure side.

6. The Doris-based real-time bin design method according to claim 1 or 3, wherein after the step of importing the Stream Load data Stream after the consuming task is executed into Doris, the method further comprises:

in an offline service scene, importing the offline service data from the HDFS into the Doris by using a Broker Load;

and configuring a scheduling task by using a Dolphin scheduler in an SQL mode, completing the processing of the offline service data in batch processing, and then summarizing the offline service data into a general table.

7. The Doris-based real-time bin design method according to claim 1 or 3, wherein after the step of importing the Stream Load data Stream after the consuming task is executed into Doris, the method further comprises:

and directly butting the relatively fixed data in the actual service scene with a data service hall, and butting the data service hall with a Power BI tool and a front-end report.

8. A Doris-based real-time bin design device is characterized by comprising:

a route Load task monitoring and setting module: the system comprises a Kakfa queue, a Topic tag, an adaptive routing Load task and a monitoring server, wherein the Kakfa queue is used for importing service real-time data into the Kakfa queue, putting the service real-time data into the adaptive topic tag, providing at least one consumption task for consuming the data in the topic tag, and setting adaptive routing Load task monitoring;

a route Load task monitoring and processing module: the offset used for monitoring and reading the position information of the current consumption data in the topic tag in real time by the Route Load task;

a Routene Load task monitoring and starting module: the method comprises the steps that if the offset does not increase within preset time, the route Load task monitoring is restarted, and the consumption task is continuously executed from the offset;

an importing module: and the Stream Load data Stream used for importing the executed consuming task into Doris.

9. A computer device, comprising:

a memory for storing a processing program;

a processor which, when executing the processing program, implements the Doris-based real-time digital bin design method of any one of claims 1 to 7.

10. A readable storage medium, having a processing program stored thereon, wherein the processing program, when executed by a processor, implements the Doris-based real-time bin design method according to any one of claims 1 to 7.

Images