CN116610752A - Transactional distributed data synchronization method, device, system and storage medium - Google Patents

Abstract

The present invention provides a transactional distributed data synchronization method, device, system and storage medium, which are used to solve the technical problem that data consistency cannot be guaranteed in the process of using Flink to synchronize data from a data source to a database table without a primary key. The present invention divides the process of Flink cluster distributed data synchronization into a pre-submission stage and a formal submission stage. In the pre-submission stage, a snapshot is created and the snapshot data is submitted to the target database in the pre-submission mode of the database transaction. After the snapshot is successfully created, the formal In the submission phase, the snapshot data is formally submitted to the destination database table. In the technical solution of the present invention, not only the processing state of the snapshot is recorded, but also the submission state of the transaction is recorded, and in the case of a Flink failure, the data synchronization task is restored according to the recorded snapshot and transaction state. The invention can realize the consistency of data synchronization of Flink from a distributed data source to a relational database table without a primary key.

Description

Transactional distributed data synchronization method, device, system and storage medium

technical field

The present invention relates to the technical field of communication and cloud computing, in particular to a transactional distributed data synchronization method, device, system and storage medium.

Background technique

Kafka is an open source stream processing platform developed by the Apache Software Foundation, written in Scala and Java. Kafka is a high-throughput distributed publish-subscribe messaging system. It is a common requirement to realize data synchronization from a distributed publish-subscribe message system (such as Kafka) to a relational database (such as MySQL).

Flink is a framework and distributed processing engine for stateful computation on unbounded (stream processing) and bounded (batch) data streams. Flink is a common tool for data synchronization from Kafka to relational databases. The general data processing process in Flink includes one or more input operator tasks (source), data processing operator tasks (transform) and output operator tasks (sink).

In the processing flow of using Flink to synchronize the data in the Kafka cluster to the relational database, due to the distributed nature of the Flink cluster and the processing logic of manual development, abnormal phenomena such as cluster restart and task rescheduling are usually prone to occur. When an exception occurs in the cluster, there will be problems with the consistency of source and destination data, such as duplicate business data in the output database, missing business data in the output database, and so on.

Through Flink's stream processing capabilities and internal mechanisms, technologies such as checkpoint snapshots can improve the consistency of data synchronization from Kafka to MySQL, especially when there is a corresponding primary key in the destination MySQL database table, which can be passed through MySQL The idempotent SQL in implements the consistency of synchronous data. However, there is no primary key defined in the MySQL database table at the destination, and when the Flink cluster is used to synchronize data from the data source Kafka to MySQL, data loss or data duplication may easily occur when the cluster is abnormal.

Contents of the invention

In view of this, the present invention provides a transactional distributed data synchronization method, device, system and storage medium, which are used to solve the problem that data consistency cannot be guaranteed in the process of using Flink to synchronize data from a data source to a database table without a primary key. technical problem.

Based on an aspect of the embodiments of the present invention, the present invention provides a transactional distributed data synchronization method, which is applied to a Flink cluster, and the method includes:

The process of synchronizing the data in the distributed data source to the destination non-primary key relational database table through the Flink cluster, that is, the destination database table, is divided into a pre-submission stage and a formal submission stage;

In the pre-submission phase, the snapshot creation and the pre-submission of the snapshot data are completed, and the output operator task sink is responsible for pre-submitting the snapshot data to the destination database table and recording the completion status of the snapshot and the pre-submission status of the transaction;

In the formal submission phase, the job manager Jobmanager is responsible for instructing the sink operator task to formally submit the snapshot data to the destination database table and record the transaction submission result after the snapshot is successfully created.

Further, the steps in the pre-submission stage include:

The job manager Jobmanager marks the currently synchronized snapshot data by injecting checkpoint boundaries;

The input operator task source records the snapshot offset after detecting the checkpoint boundary;

After the sink operator task detects the checkpoint boundary, it is responsible for caching the current snapshot data, pre-submitting the current snapshot data to the destination database table, and recording the pre-commit status of the transaction;

Each operator task sends a snapshot completion feedback to the Jobmanager after successfully completing the processing of the snapshot data, and the sink operator task sends a pre-submission result feedback to the Jobmanager after successfully completing the pre-submission processing of the snapshot data;

After the Jobmanager receives the snapshot completion feedback of all operator tasks and the pre-submission result feedback of the sink operator task, it records the snapshot success status.

Further, the steps in the formal submission stage include:

After the Jobmanager judges that the creation and pre-submission are successful based on the snapshot completion feedback and pre-submission result feedback, it sends a snapshot completion message to all operator tasks;

After receiving the snapshot completion message, the output operator task sink formally submits the snapshot data to the destination database table and records the transaction submission result of whether the transaction is formally submitted successfully or not.

Further, if the Flink cluster fails during the pre-submit phase and causes the snapshot to be created unsuccessfully, the method also includes the following fault recovery steps:

Jobmanager obtains the snapshot information of the last successfully created snapshot from the state storage backend;

Jobmanager coordinates and schedules Taskmanager to restore data synchronization tasks;

The sink operator task obtains the transaction identification ID and snapshot data of the last successfully created snapshot from the state storage backend according to the snapshot information;

The sink operator task obtains the transaction submission status according to the transaction ID, and verifies whether the snapshot data of the latest snapshot successfully created is successfully submitted to the destination database table. If the submission is successful, the source operator task obtains the next snapshot to be consumed Data offset, restart the data synchronization process.

Further, if the Flink cluster fails during the pre-submit phase and the latest snapshot is created successfully, the method also includes the following fault recovery steps:

Jobmanager obtains the snapshot information of the latest snapshot successfully created;

Jobmanager coordinates and schedules Taskmanager to restore the execution of data synchronization tasks;

The sink operator task obtains the transaction identification ID and snapshot data of the last successfully created snapshot according to the snapshot information;

The sink operator task obtains the transaction submission status according to the transaction ID, and judges whether the corresponding snapshot data is formally submitted to the destination database table;

In the state that the transaction is not officially committed, the sink operator task submits the snapshot data of the last successfully created snapshot to the destination database table;

The source operator task obtains the data offset of the next snapshot to be consumed to continue data synchronization.

Further, the Jobmanager and each operator task record snapshot information in the state storage backend, the state storage backend is a distributed file system; the distributed data source is a distributed publish-subscribe message system; the sink Operator tasks record transaction status in a distributed memory database.

Based on another aspect of the embodiments of the present invention, the present invention also provides a transactional distributed data synchronization system, which includes a Flink cluster, a state storage backend, and a distributed memory database;

The Flink cluster includes a job manager Jobmanager and a task manager Taskmanager, and the Taskmanager runs at least an input operator task source and an output operator task sink;

The Flink cluster is used to synchronize the data in the distributed data source to the target non-primary key relational database table, that is, the target database table, and the data synchronization process is divided into a pre-submission stage and a formal submission stage;

In the pre-submission phase, the creation of snapshots and the pre-submission of snapshot data are completed, and the output operator sink operator task is responsible for pre-submitting the snapshot data to the destination database table and recording the completion status of the snapshot and the pre-submission status of the transaction;

In the formal submission phase, the job manager Jobmanager is responsible for instructing the sink operator task to formally submit the snapshot data to the destination database table and record the transaction submission results after the snapshot is successfully created;

The state storage backend is used to store snapshot information and cache snapshot data;

The distributed memory database is used to store transaction status.

Further, in the pre-submit phase:

The Jobmanager marks the currently synchronized snapshot data by injecting checkpoint boundaries;

The source operator task records the snapshot offset after detecting the checkpoint boundary;

After the sink operator task detects the checkpoint boundary, it is responsible for caching the current snapshot data, pre-submitting the current snapshot data to the destination database table, and recording the transaction pre-submission status;

Each operator task sends a snapshot completion feedback to the Jobmanager after successfully completing the processing of the snapshot data, and the sink operator task sends a pre-submission result feedback to the Jobmanager after successfully completing the pre-submission processing of the snapshot data;

The Jobmanager is also used to receive the snapshot completion feedback of all operator tasks and the pre-submission result feedback of the sink operator task, and record the success status of the snapshot.

Further, in the formal submission stage:

The Jobmanager is also used to send a snapshot completion message to all operator tasks after the snapshot completion feedback and the pre-submission result feedback determine that the creation is successful and the pre-submission is successful;

After receiving the snapshot completion message, the output operator task sink formally submits the snapshot data to the target database table and records the transaction submission result of whether the transaction formally submits successfully or not.

Based on one aspect of the embodiments of the present invention, the present invention also provides a transactional distributed data synchronization device, which is used to synchronize the data in the distributed data source to the target non-primary key relational database table, that is, the target database table through the Flink cluster , the device includes:

The pre-submission module is used to complete the processing steps of the pre-submission stage in the data synchronization process, complete the snapshot creation and the pre-submission of the snapshot data; wherein, the snapshot data is pre-submitted to the destination database table through the output operator task sink and Record snapshot completion status and transaction pre-commit status;

The formal submission module is used to complete the processing steps of the formal submission phase in the data synchronization process, and submit the snapshot data to the destination database table formally in the transaction mode; wherein, after the snapshot is created successfully, the job manager Jobmanager instructs the sink to calculate The subtask formally submits the snapshot data to the target database table and records the transaction submission status and transaction submission results.

The device provided by the present invention can be implemented in the form of software, hardware or a combination of software and hardware. When implemented as a software module, the program code of the software module is loaded into the storage medium of the device, and the processor reads and executes the program code in the storage medium, thereby realizing the functions of each component module in the device.

Based on an aspect of the embodiments of the present invention, the present invention also provides an electronic device, including a processor, a communication interface, a storage medium, and a communication bus, wherein, the processor, the communication interface, and the storage medium communicate with each other through the communication bus;

storage medium for storing computer programs;

The processor is configured to implement one or more steps in the foregoing transactional distributed data synchronization method when executing the computer program stored on the storage medium.

The invention provides a transactional distributed data synchronization method, device, system and storage medium, which are used to solve the technical problem that data consistency cannot be guaranteed in the process of using Flink to synchronize data from a data source to a database table without a primary key. The present invention divides the process of Flink cluster distributed data synchronization into a pre-submission stage and a formal submission stage. In the pre-submission stage, a snapshot is created and the snapshot data is submitted to the target database in the pre-submission mode of the database transaction. After the snapshot is successfully created, the formal In the submit phase, the snapshot data is formally submitted to the destination database table. In the technical solution of the present invention, not only the processing state of the snapshot is recorded, but also the submission state of the transaction is recorded, and in the case of a Flink failure, the data synchronization task is restored according to the recorded snapshot and transaction state. The invention can realize the consistency of data synchronization of Flink from a distributed data source to a relational database table without a primary key.

Description of drawings

In order to more clearly illustrate the technical solutions in the embodiments of the present invention or the prior art, the following will briefly introduce the accompanying drawings that need to be used in the description of the embodiments of the present invention or the prior art. Obviously, the accompanying drawings in the following description These are only some embodiments described in the present invention, and those skilled in the art can also obtain other drawings according to these drawings of the embodiments of the present invention.

Fig. 1 is a schematic diagram of the process of realizing data synchronization based on the Flink two-stage distributed transaction state mechanism in an embodiment of the present invention;

FIG. 2A is a schematic diagram of initiating pre-submission and executing a pre-submission process in the pre-submission stage in an embodiment of the present invention;

FIG. 2B is a schematic diagram of confirming the completion of the pre-submission phase in the pre-submission phase in an embodiment of the present invention;

FIG. 2C is a schematic diagram of executing the second stage, namely the formal submission stage, in an embodiment of the present invention;

FIG. 3A is a schematic diagram of abnormal recovery steps in the case of unsuccessful snapshot creation in the first stage in an embodiment of the present invention;

FIG. 3B is a schematic diagram of abnormal recovery steps in the case of successful snapshot creation in the first stage in an embodiment of the present invention;

FIG. 4 is a schematic structural diagram of an electronic device for implementing a method for synchronizing transactional distributed data according to an embodiment of the present invention.

Detailed ways

The terms used in the embodiments of the present invention are only for the purpose of describing specific embodiments, rather than limiting the embodiments of the present invention. The singular forms "a", "said" and "the" used in the embodiments of the present invention are also intended to include plural forms unless the context clearly indicates otherwise. It should be understood that although the terms first, second, third, etc. may be used in the embodiments of the present invention to describe various information, the information should not be limited to these terms. These terms are only used to distinguish similar information, entities or steps, and are not used to describe a specific order or sequence. For example, without departing from the scope of the embodiments of the present invention, first information may also be called second information, and similarly, second information may also be called first information. Additionally, the use of the word "if" may be construed as "at" or "when" or "in response to a determination." The "and/or" in the present invention is only an association relationship describing associated objects, which means that there can be three relationships, for example, A and/or B, which can mean: A exists alone, A and B exist simultaneously, and there exists alone B these three cases, where A, B can be singular or plural. And, in the description of the present invention, unless otherwise specified, "plurality" means two or more than two. "At least one of the following" or similar expressions refer to any combination of these items, including any combination of single or plural items. For example, at least one item (piece) of a, b, or c can represent: a, b, c, a-b, a-c, b-c, or a-b-c, where a, b, c can be single or multiple .

Flink is a distributed stream processing framework that can execute arbitrary stream data programs in a parallel and pipeline manner. Flink's pipeline runtime system can execute batch and stream processing programs. Flink's checkpoint Checkpoint snapshot mechanism is the cornerstone of its reliability. When a task fails during operation, it can restore to the state before the failure according to the Checkpoint snapshot information, and then resume the operation of the task from this state. In Flink, the Checkpoint snapshot mechanism adopts a distributed snapshot algorithm. Through the Checkpoint snapshot mechanism, the semantics of exactly once (Exactly Once) inside the Flink program are guaranteed, and the internal data consistency of Flink is guaranteed.

The Flink framework includes two important components, namely the management and coordination component JobManager (job manager) and the execution process component TaskManager (task manager). The Flink architecture also follows the Master-Slave master-slave architecture design principle. The JobManager is equivalent to the entire cluster. Master node, and the entire cluster has one and only one active JobManager, TaskManager is equivalent to the Slave node of the entire cluster.

JobManager is responsible for the scheduling of the entire Flink cluster task and resource management, and TaskManager is responsible for specific task execution and resource application and management of corresponding tasks on each node. The JobManager obtains the submitted application from the client, and then allocates the corresponding TaskSlot resources for the submitted application according to the usage of the task slot TaskSlot on the TaskManager in the cluster and instructs the TaskManager to start the application obtained from the client.

The client compiles and packages the written Flink application, submits it to the JobManager, and then the JobManager assigns the task to the TaskManager node with resources according to the resources of the TaskManager registered in the JobManager, and then starts and runs the task.

In the scenario where Flink is used to synchronize data from Kafka to relational databases, based on the Flink framework's native snapshot checkpoint Checkpoint mechanism, a "state storage backend" is added to store the snapshot bias of the business data processed by the source operator task in the Flink cluster. Shift offset, the snapshot offset is the starting position of the data consumed by the source operator task from Kafka, the sink operator task is connected to the database, and writes the processed business data output into the database table. Each time the processing of a batch of business data corresponding to the snapshot is completed, Jobmanager will write the snapshot offset offset to the backend of the state storage. When a cluster or task is abnormal, Flink will restore to the state when the last snapshot was successful, and proceed to the next step A snapshot data processing, but the realization process will face the following two problems:

Question 1: If this snapshot is not successful, but some data has been written into the database table, an exception occurs in the Flink cluster at this time, the task restarts, and it will be restored from the previous snapshot again. At this time, the data that has been stored in the database may be processed repeatedly , causing duplicate data to be inserted into the database table.

Question 2: If the snapshot is successful, but the snapshot data is not completely written into the database, an exception occurs in the Flink cluster. When the task is restarted, it will continue to process the next snapshot data from the last successful snapshot. In this case In this case, some business data will be lost because some snapshot data is not stored in the database.

It can be seen that based on Flink's native checkpoint snapshot mechanism, in the above cases, only the data consistency between the data source and Flink can be guaranteed, but for the sink output, if the destination database table has no primary key, the data source cannot be guaranteed The consistency of business data in the end and destination database tables.

In order to solve the technical problem that data consistency cannot be guaranteed in the process of using Flink to synchronize data from a data source to a database table without a primary key, the present invention proposes a transactional distributed data synchronization scheme. The core idea of the technical solution of the present invention is to divide the process of using the Flink cluster for distributed data synchronization into a pre-submission stage and a formal submission stage. In the pre-submission stage, the sink operator task submits the snapshot data to the destination in the pre-submission mode of the database transaction. For the database, the JobManager records the snapshot success status after receiving the snapshot success feedback of all operator tasks and notifies the sink operator task to formally submit the snapshot data to the destination database table, and the sink operator task records the transaction submission status after submitting the data successfully. In the technical solution of the present invention, not only the processing state of the snapshot is recorded, but also the submission state of the transaction is recorded, and in the case of a Flink failure, the data synchronization task is restored according to the recorded snapshot and transaction state. The present invention can realize the consistency of data synchronization of Flink from a distributed data source (such as a distributed publish-subscribe message system, Kafka, etc.) to a relational database (such as MySQL) table without a primary key.

Based on the basic idea of the present invention, the specific implementation process of the present invention will be described below in conjunction with the drawings and specific embodiments.

Fig. 1 is a schematic diagram of the process of realizing data synchronization based on Flink's two-stage distributed transaction state mechanism in an embodiment of the present invention. This embodiment uses Flink to divide the data synchronization process based on the distributed transaction state mechanism from the distributed data source to the non-primary key relational database table into two stages, the first stage is the pre-commit stage, and the second stage is the formal commit stage. The coordination management component JobManager (job manager) in the Flink cluster acts as a coordinator to coordinate the creation of snapshots and the submission of transactions in two phases, and each operator task managed by the execution process component TaskManager (task manager) in the Flink cluster As a participant is responsible for data synchronization (data acquisition, data processing, data output). After completing the pre-commit phase of the first phase, the JobManager initiates the processing steps of the second phase. JobManager and TaskManager cooperate to complete distributed data synchronization based on transaction mechanism.

Through the two-stage transactional writing process, the present invention can not only ensure the consistency of synchronous data inside Flink but also ensure the data consistency of Sink output end, thereby ensuring the consistency of data from data sources to non-primary key relational database tables.

The processing procedures of the two stages are described in detail below.

FIG. 2A is a schematic diagram of initiating pre-submission and executing the pre-submission process in the pre-submission phase in an embodiment of the present invention; FIG. 2B is a schematic diagram of confirming completion of the pre-submission phase in an embodiment of the present invention. The data source for data synchronization in the figure is Kafka. The present invention does not limit the type and specific form of the data source. For example, it can also be other distributed message subscription and publishing components, or other data collection systems. The destination of the data synchronization in the figure is a MySQL database table without primary key index (no primary key). The present invention does not limit the type of the target relational database, such as MySQL, SQLServer, Oracle, etc. In the figure, JobManager and TaskManager are the components of the Flink engine itself. The distributed file system HDFS is used as the state storage backend to cache snapshot offsets and snapshot data. The distributed memory database Redis is a newly introduced component to cache transaction status information.

The first stage of processing based on Figure 2A and Figure 2B includes:

Step 200, the job manager in the Flink cluster injects the checkpoint boundary into the data input operator Source, triggering the snapshot creation process;

The job manager Jobmanager in the Flink cluster is responsible for scheduling and executing data synchronization jobs. After the client submits the data synchronization job to the Jobmanager, the Jobmanager schedules the job tasks of the data synchronization job to the operators (Source, Map, Sink) of the task manager Taskmanager. To run these operator tasks.

Flink itself provides the Checkpoint checkpoint snapshot function, which is implemented by injecting the checkpoint boundary line. The checkpoint boundary line can be regarded as a snapshot mark, and its main function is to distinguish between two checkpoint checkpoint snapshots before and after. A snapshot Corresponds to a batch of business data that can be processed by multiple task operators. The checkpoint boundary is a special data flow triggered by the Jobmanager component. The Jobmanager injects the checkpoint boundary from the data input operator (source) task. The boundary will follow the entire data processing process and the normal data flow. Process flow, so each operator will detect the checkpoint boundary and process accordingly.

Step 201, during the execution of the Flink cluster data input operator task source, after detecting the checkpoint boundary, record the snapshot offset;

The data input source operator task consumes data from Kafka. After receiving the instruction of Jobmanager injecting the checkpoint boundary line, it takes a snapshot of the consumed data and offsets the snapshot corresponding to the checkpoint boundary line (that is, the currently consumed The location information of kafka data) is recorded to the state storage backend.

The state storage backend used in the embodiment of the present invention may be a distributed file system (such as Hadoop Distributed File System, HDFS), or other types of distributed file systems or databases, which is not limited in the present invention.

Step 202, the data output operator task sink in the Flink cluster detects the checkpoint boundary, starts the transaction, and caches the snapshot data and transaction ID;

There may be multiple sink operator tasks in the Flink cluster to cooperate to complete the data synchronization job. After each sink operator task detects the checkpoint boundary, it starts a database transaction for the current snapshot data and marks the checkpoint boundary. The snapshot data is serialized, and the snapshot data and transaction ID are serialized and cached to the state storage backend for use when the cluster fails to recover. The sink operator task can also use other storage locations to cache the snapshot data of the serial number, which is not stored together with the snapshot status information such as the snapshot offset.

Step 203, the sink operator task pre-submits the current snapshot data to the destination database table and records the pre-commit status of the transaction;

There may be multiple sink operator tasks in the Flink cluster to cooperate to complete the data synchronization job. After each sink operator task detects the checkpoint boundary line, it will start a transaction for the current snapshot data, and the snapshot marked by the checkpoint boundary line After the data and transaction ID are serialized and cached to the state storage backend, the snapshot data is pre-submitted to the destination database. After the snapshot data pre-submission is completed, the sink operator task also needs to record the snapshot data in the distributed memory database redis The pre-commit status of the snapshot data is marked as the pre-commit state (uncommitted mode), and the transaction ID and the success or failure result of the pre-commit need to be recorded.

The transaction in the present invention is consistent with the concept of the transaction in the database. When starting a transaction in the database, before executing the Commit command, the addition/deletion/modification of the database table will not actually take effect, and the Rollback command can Roll back the current transaction to the state before the transaction starts at any time. The processing process of the pre-commit phase in the present invention is equivalent to the processing process before executing the Commit instruction in the database transaction.

In the present invention, the Sink operator task writes the snapshot data into the target database in the pre-write mode of the database transaction. Although this part of the snapshot data has actually been written into the database, it is in the state of not being officially committed to the target database table. Only after all the sink operator tasks have completed the pre-submission of all the snapshot data of the current batch and recorded the transaction pre-commit status, the Jobmanager can further trigger the sink operator tasks to perform the Commit operation that formally submits the snapshot data to the destination database table .

Step 204, each operator task in the Flink cluster sends snapshot completion feedback and pre-submission result feedback to the management coordination component after completing the pre-submission phase processing of the snapshot data they are responsible for;

Step 205. After receiving the snapshot completion feedback and pre-submission result feedback from all operator task feedbacks, the management coordination component records the snapshot success status of the injected checkpoint boundary mark;

This step corresponds to the example in Figure 2B. After each operator task completes the processing of the pre-submission phase related to the current snapshot, it will send snapshot completion feedback and pre-submission result feedback to the management coordination component Jobmanager.

For example, after the source operator task finishes recording the snapshot offset to the state storage backend, it will send a snapshot completion feedback to the Jobmanager; after the sink operator task completes the snapshot data cache to the state storage backend, it will also send a snapshot to the Jobmanager Completion of the feedback, after the sink operator task completes the pre-submission of the snapshot data and records the pre-submission status, it will also send the pre-submission result feedback to the Jobmanager.

The management and coordination component Jobmanager receives feedback from all operators about the current snapshot and pre-submission status, and after judging that all operator tasks have successfully completed their respective processing steps according to the feedback results, records the successful status of the snapshot to the state storage backend, and the snapshot is successful. The state information may include a checkpoint boundary mark, a snapshot ID, a snapshot time, and the like. Jobmanager can confirm the completion of the first phase only after each operator task successfully completes the snapshot creation and the Sink operator task successfully completes the pre-submission of synchronous data and records the transaction status. If any operator task fails to successfully complete the snapshot creation or the pre-commit of the snapshot data, it means that the processing of the first phase of the current snapshot, that is, the pre-commit phase, fails, which leads to the failure of the data synchronization of the two-phase transaction status of the current snapshot. At this time, you need to execute Corresponding troubleshooting steps.

In some embodiments of the present invention, there are generally source and sink operator tasks. Normally, there may be multiple Map data processing operators between the source and the sink for processing data synchronization. Map operator tasks are generally stateless operators that do not need to record state information.

FIG. 2C is a schematic diagram of executing the second stage, that is, the formal submission stage, in an embodiment of the present invention. The process of executing the second stage based on FIG. 2C includes:

Step 206, the job manager sends a snapshot completion message to all operator tasks;

After the job manager Jobmanager writes the status of successful snapshot creation to the state storage backend, it sends a snapshot completion message to all operators (Source, Map, Sink) to indicate the completion of the task pre-submission phase of each operator.

Step 207, after receiving the snapshot completion message, the sink operator task officially submits the current snapshot data to the destination database table and records the transaction submission status as the submission completion status;

When the sink operator task receives the notification message of the completion of the pre-commit phase, it officially submits the snapshot data Commit it is responsible for to the target database table, and after the commit is successful, it records the status information of the successful commit of the transaction to the distributed memory database redis. The snapshot data corresponding to the injected checkpoint boundary line is successfully synchronized to the target relational database table, and the second-stage process based on the distributed transaction state mechanism is successfully completed.

An important technical effect of the two-stage distributed data synchronization based on the transaction mechanism in the present invention is that it can achieve data consistency from Kafka to the destination database in the event of a Flink cluster failure. The normal feedback about the current snapshot from all operators was not received within a certain period of time, some operators reported that snapshot creation failed, sink operator task feedback failed to pre-submit, etc. The snapshot data synchronization failure marked by the checkpoint boundary line may occur in the pre-commit phase or in the formal commit phase. The following describes the fault recovery methods for the two cases.

Abnormal situation 1: When the data synchronization job process is interrupted due to a failure, it is in the first stage of pre-submission stage, and the checkpoint snapshot is not successful.

Whether the snapshot creation is successful is judged by whether the Jobmanager receives the snapshot completion feedback message sent by all operator tasks indicating that the snapshot is created successfully. If the snapshot completion feedback of all operator tasks is received, the Jobmanager determines that the snapshot creation is successful. The snapshot creation success status record information will be recorded to the state storage backend. In some cases, the Map operator task is stateless and does not need to send a snapshot completion feedback message to the Jobmanager. The Jobmanager mainly determines whether the snapshot completion feedback sent by all source and sink operator tasks is received to determine whether the snapshot creation is successful.

Figure 3A is a schematic diagram of the abnormal recovery steps in the case of unsuccessful creation of the snapshot in the first stage in an embodiment of the present invention. If the Flink cluster fails before the Jobmanager records the status record information of the successful snapshot creation, the execution is as follows after the cluster is restarted step:

Step 310, the Jobmanager obtains the snapshot information of the last successfully created snapshot (the previous successful snapshot of the latest failed snapshot) from the state storage backend;

The snapshot information of the last successfully created snapshot may include corresponding checkpoint boundary information (snapshot identifier), snapshot offset, snapshot creation time, and the like.

Step 311, Jobmanager coordinates and schedules Taskmanager to restore the execution of the data synchronization task;

Step 312, the sink operator task obtains the transaction ID and the snapshot data of the last successfully created snapshot from the state storage backend according to the snapshot information;

Step 313, the sink operator task obtains the transaction submission status according to the transaction ID, and verifies that the transaction submission status is submitted;

This step may include the sink operator task obtaining the cached serialized snapshot data from the state storage backend, recovering the transaction ID and snapshot data through deserialization processing, and then judging from redis the latest successfully created snapshot corresponding to the transaction ID Whether the transaction submission is successful, if successful, continue to execute the next steps, if not, you need to submit the corresponding snapshot data to the destination database table, and then execute the next steps.

Step 314, the source operator task obtains the offset of the next snapshot data to be consumed (the latest failed snapshot data) from the state storage backend, and restarts the data synchronization process such as creating a snapshot.

Figure 3B is a schematic diagram of the abnormal recovery steps when the first-stage snapshot is successfully created in an embodiment of the present invention. If the Flink cluster fails after the Jobmanager records that the latest snapshot is successfully created, there may be some sink operator tasks in this case If the committed transaction is successful, but the second phase has not been successfully completed, perform the following steps after the cluster restarts:

Step 320, the Jobmanager obtains the snapshot information of the last successfully created snapshot from the state storage backend;

Step 321, Jobmanager coordinates and schedules the execution of Taskmanager recovery data synchronization task;

Step 322, the sink operator task obtains the transaction ID and the snapshot data of the last successfully created snapshot from the state storage backend according to the snapshot information;

Step 323, the sink operator task obtains the transaction submission status according to the transaction ID;

Step 324, in the state of transaction uncommitted (snapshot data has not been formally submitted in the destination database table), the sink operator task submits the recovered uncommitted transaction to the destination database table;

Step 325, the source operator task obtains the next starting point offset (offset of different data blocks of the same snapshot) to be consumed from the state storage backend (through deserialization operation) to continue data synchronization.

A snapshot marked by a checkpoint boundary corresponds to a batch of data in Kafka. Multiple sources can consume this batch of data in Kafka in parallel. A source operator task may process multiple data in this batch of data belonging to the same snapshot. For data blocks, each operator task can identify a snapshot according to the checkpoint boundary, and identify which data block is being processed through the data block identifier.

If the Flink cluster fails in the second stage of the snapshot data synchronization process, it means that the snapshot creation has been successful. Therefore, the processing process is similar to the process in Figure 3B. The sink operator task judges whether the transaction submission is successful based on the transaction ID. If not, then Formally submit the transaction, and if successful, continue to consume the same batch of snapshot data in Kafka to perform data synchronization.

4 is a schematic structural diagram of an electronic device for implementing the transactional distributed data synchronization method provided by the present invention according to an embodiment of the present invention. The device 400 includes: a processor 410 such as a central processing unit (CPU), a communication bus 420 , a communication interface 440 and a memory 430. Wherein, the processor 410 and the memory 430 may communicate with each other through the communication bus 420 . A computer program is stored in the memory 430, and when the computer program is executed by the processor 410, the functions of one or more steps in the transactional distributed data synchronization method provided by the present invention can be realized.

Memory refers to a device based on a certain storage medium for storing computer programs and/or data. It can be a volatile memory (Volatile Memory, VM, often called memory) or a non-volatile memory (Non-Volatile Memory, NVM). Memory refers to the internal memory that directly exchanges data with the processor. It can read and write data at any time, and the speed is very fast. It is used as a storage medium for temporary data of the operating system and other running programs. The memory may be Synchronous Dynamic Random Access Memory (Synchronous Dynamic Random Access Memory, SDRAM), Dynamic Random Access Memory (Dynamic Random Access Memory, DRAM), etc. Non-volatile memory refers to a memory using a persistent storage medium, which has the characteristics of large capacity and persistent data storage. It can be storage class memory (StorageClass Memory, SCM), solid state disk (Solid State Disk, SSD), NAND flash memory, disk etc. SCM is the industry's general term for new storage media between memory and flash memory. It is a composite storage technology that combines persistent storage features and memory features. The access speed is slower than DRAM and faster than SSD hard drives.

The processor can be a general-purpose processor, including a central processing unit (Central Processing Unit, CPU), a network processor (Network Processor, NP), etc.; it can also be a digital signal processor (Digital Signal Processing, DSP), an application-specific integrated circuit (Application Specific Integrated Circuit, ASIC), Field-Programmable Gate Array (Field-Programmable Gate Array, FPGA) or other programmable logic devices, discrete gate or transistor logic devices, discrete hardware components.

It should be appreciated that embodiments of the present invention may be realized or implemented by computer hardware, a combination of hardware and software, or by computer instructions stored in non-transitory (or referred to as non-persistent) memory. The methods can be implemented in a computer program using standard programming techniques, including a non-transitory storage medium configured with a computer program, wherein the storage medium so configured causes the computer to operate in a specific and predefined manner. Each program can be implemented in a high-level procedural or object-oriented programming language to communicate with the computer system. However, the programs can be implemented in assembly or machine language, if desired. In any case, the language may be a compiled or interpreted language. Furthermore, the program can be run on an application specific integrated circuit programmed for this purpose. Furthermore, operations of processes described herein may be performed in any suitable order unless otherwise indicated herein or otherwise clearly contradicted by context. The processes described herein (or variations and/or combinations thereof) can be performed under the control of one or more computer systems configured with executable instructions and as code that is commonly executed on one or more processors ( For example, executable instructions, one or more computer programs or one or more applications), by hardware or a combination thereof. The computer program comprises a plurality of instructions executable by one or more processors.

Further, the method can be implemented in any type of computing platform operably connected to a suitable one, including but not limited to personal computer, minicomputer, main frame, workstation, network or distributed computing environment, stand-alone or integrated computer platform, or communicate with charged particle tools or other imaging devices, etc. Aspects of the invention can be implemented as machine-readable code stored on a non-transitory storage medium or device, whether removable or integrated into a computing platform, such as a hard disk, optically read and/or written storage medium, RAM, ROM, etc., such that they are readable by a programmable computer, when the storage medium or device is read by the computer, can be used to configure and operate the computer to perform the processes described herein. Additionally, the machine-readable code, or portions thereof, may be transmitted over a wired or wireless network. The invention described herein includes these and other various types of non-transitory computer-readable storage media when such media include instructions or programs that implement the steps described above in conjunction with a microprocessor or other data processor. The invention also includes the computer itself when programmed according to the methods and techniques described herein.

The above descriptions are only examples of the present invention, and are not intended to limit the present invention. Various modifications and variations of the present invention will occur to those skilled in the art. Any modification, equivalent replacement, improvement, etc. made within the spirit and principle of the present invention shall be included in the protection scope of the present invention.

Claims (13)

1. A method for transactional distributed data synchronization, the method being applied to a Flink cluster, the method comprising:

the method comprises the steps that a flank cluster is used for synchronizing data in a distributed data source to a target keyless relational database table, namely a target database table, and the process is divided into a pre-submitting stage and a formal submitting stage;

in the pre-commit stage, snapshot creation and pre-commit of snapshot data are completed, and an output operator task sink is responsible for pre-commit of the snapshot data to a target database table and records a snapshot completion state and a transaction pre-commit state;

in the formal commit phase, the job manager Jobmanager is responsible for instructing sink operator tasks to formally commit snapshot data to the destination database table after the snapshot is created successfully and recording transaction commit results.

2. The method of claim 1, wherein the pre-commit phase comprises:

The job manager Jobmanager marks the currently synchronized snapshot data by injecting checkpoint boundaries;

the input operator task source records snapshot offsets after detecting checkpoint boundaries;

after detecting a check point boundary, the sink operator task is responsible for caching current snapshot data and pre-submitting the current snapshot data to a target database table and recording a transaction pre-submitting state;

each operator task sends snapshot completion feedback to the Jobmanager after the snapshot data is successfully processed, and the sink operator task sends pre-submission result feedback to the Jobmanager after the snapshot data is successfully submitted;

and after the Jobmanager receives snapshot completion feedback of all operator tasks and pre-submitting result feedback of sink operator tasks, recording a snapshot success state.

3. The method of claim 2, wherein the step of formally submitting the phase comprises:

after Jobmanager judges that creation is successful and pre-submission is successful according to snapshot completion feedback and pre-submission result feedback, snapshot completion information is sent to all operator tasks;

and after receiving the snapshot completion message, the output operator task sink formally submits snapshot data to a target database table and records a transaction submitting result whether the transaction formally submits successfully or not.

4. The method of claim 2, wherein if the Flink cluster fails during the pre-commit phase and results in unsuccessful snapshot creation, the method further comprises the step of failure recovery:

jobmanager obtains snapshot information of a snapshot which is successfully created last time from a state storage back end;

jobmanager coordinates the task manager to resume the data synchronization task;

acquiring a transaction identification ID and snapshot data of a snapshot which is successfully created last time from a state storage back end according to the snapshot information by a sink operator task;

and the sink operator task acquires a transaction submitting state according to the transaction identification ID, verifies whether the snapshot data of the snapshot which is successfully created in the last time is successfully submitted to a target database table, and if so, the source operator task acquires the snapshot data offset to be consumed next time, and resumes the data synchronization process.

5. The method of claim 3, wherein if the flank cluster failed during the pre-commit phase and the last snapshot creation was successful, the method further comprises the step of recovering from the failure:

jobmanager obtains snapshot information of a snapshot which is successfully created last time;

jobmanager coordinates the task manager to resume execution of the data synchronization task;

Acquiring a transaction identification ID and snapshot data of a snapshot which is successfully created last time according to the snapshot information by a sink operator task;

the sink operator task acquires the transaction submitting state according to the transaction ID and judges whether the corresponding snapshot data is formally submitted to the target database table;

under the condition that the transaction is not formally submitted, the sink operator task submits snapshot data of the snapshot which is successfully created last time to a target database table;

the source operator task acquires the next snapshot data offset to be consumed and continues to perform data synchronization.

6. The method of claim 1, wherein the step of determining the position of the substrate comprises,

the Jobmanager and each operator task record snapshot information in a state storage back end, wherein the state storage back end is a distributed file system;

the distributed data source is a distributed publish-subscribe messaging system.

7. The method of claim 1, wherein the step of determining the position of the substrate comprises,

the sink operator task records the transaction state in a distributed memory database.

8. The transactional distributed data synchronization system is characterized by comprising a Flink cluster, a state storage back end and a distributed memory database;

the Flink cluster comprises a job manager Jobmanager and a task manager Taskanager, wherein at least an input operator task source and an output operator task sink are operated in the task manager Taskanager;

The Flink cluster is used for synchronizing data in the distributed data source into a target database table which is a target database table without a main key relation, and the data synchronization process is divided into a pre-submission stage and a formal submission stage;

in the pre-submitting stage, snapshot creation and pre-submitting of snapshot data are completed, and an output operator sink operator task is responsible for pre-submitting the snapshot data to a target database table and recording a snapshot completion state and a transaction pre-submitting state;

in the formal submitting stage, the job manager Jobmanager is responsible for instructing a sink operator task to formally submit snapshot data to a target database table after the snapshot is successfully created and recording a transaction submitting result;

the state storage back end is used for storing snapshot information and caching snapshot data;

the distributed memory database is used for storing transaction states.

9. The system of claim 8, wherein in the pre-commit phase:

the Jobmanager marks currently synchronized snapshot data through injecting check point boundaries;

the source operator task records snapshot offsets after detecting a checkpoint boundary;

after detecting a check point boundary, the sink operator task is responsible for caching current snapshot data and pre-submitting the current snapshot data to a target database table and recording a transaction pre-submitting state;

Each operator task sends snapshot completion feedback to the Jobmanager after the snapshot data is successfully processed, and the sink operator task sends pre-submission result feedback to the Jobmanager after the snapshot data is successfully submitted;

the Jobmanager is also used for receiving snapshot completion feedback of all operator tasks and pre-submitting result feedback of sink operator tasks and recording snapshot success states.

10. The system of claim 9, wherein in the formal commit phase:

the Jobmanager is also used for sending snapshot completion information to all operator tasks after the creation success and the pre-submission success are judged according to snapshot completion feedback and pre-submission result feedback;

and after receiving the snapshot completion message, the output operator task sink formally submits snapshot data to a target database table and records a transaction submission result of whether the transaction formally submits successfully or not.

11. A transactional distributed data synchronization apparatus for synchronizing data in a distributed data source to a destination keyless relational database table, i.e., a destination database table, via a flank cluster, the apparatus comprising:

the pre-submitting module is used for completing the processing steps of the pre-submitting stage in the data synchronization process and completing snapshot creation and pre-submitting of snapshot data; the method comprises the steps of pre-submitting snapshot data to a target database table through an output operator task sink and recording a snapshot completion state and a transaction pre-submitting state;

The formal submitting module is used for completing the processing steps of the formal submitting stage in the data synchronization process and formally submitting the snapshot data in a target database table in a transaction mode; and after the snapshot is successfully created, the job manager Jobmanager instructs the sink operator task to formally submit snapshot data to the target database table, record the transaction submission state and record the transaction submission result.

12. An electronic device is characterized by comprising a processor, a communication interface, a storage medium and a communication bus, wherein the processor, the communication interface and the storage medium are communicated with each other through the communication bus;

a storage medium storing a computer program;

a processor for carrying out the method steps of any one of claims 1-7 when executing a computer program stored on a storage medium.

13. A storage medium having stored thereon a computer program which, when executed by a processor, implements the method of any of claims 1 to 7.