What's New
Function Overview
Product Bulletin
- Product Notice
- Version Support Notes
Service Overview
- GaussDB(DWS) Infographics
- What Is GaussDB(DWS)?
- Data Warehouse Types
- Data Warehouse Flavors
- Advantages
- Application Scenarios
- Functions
- Concepts
- Related Services
- Security
- GaussDB(DWS) Permissions Management
- GaussDB(DWS) Access
- Restrictions
- Technical Support
- Service Quotas
- GaussDB(DWS) Technical Specifications
Billing
- GaussDB(DWS) Billing Overview
- Billing Modes
- Item
- Billing Examples
- Billing Mode Change
- Renewal
  - Overview
  - Manual Renewal
- Bills
- Arrears
- Stopping Billing
- Cost Management
- Billing FAQs
Getting Started
- Checkpoint Vehicle Analysis
- Supply Chain Requirement Analysis of a Company
- Operations Status Analysis of a Retail Department Store
- Creating a Time Series Table
- Best Practices of Hot and Cold Data Management
- Best Practices for Automatic Partition Management
- Creating a Cluster and Connecting to It
- Using CDM to Migrate MySQL Data to the GaussDB(DWS) Cluster
- Using DLI Flink Jobs to Write Kafka Data to GaussDB(DWS) in Real Time
- Basic SQL Operations
- Database Quick Start
- Getting Started with Common Practices
User Guide
- Using GaussDB(DWS)
- Preparations
- Creating a GaussDB(DWS) Cluster
- Connecting to a GaussDB(DWS) Cluster
- Creating a GaussDB(DWS) Database and User
- Migrating Service Data to a GaussDB(DWS) Cluster
  - Migrating Data to a GaussDB(DWS) Cluster Using GDS-Kafka
  - Data Source Management
- GaussDB(DWS) Cluster Data Security and Encryption
- GaussDB(DWS) Cluster Management
- GaussDB(DWS) Cluster O&M
Best Practices
- Import and Export
- Data Migration
- Data Analytics
- Decoupled Storage and Compute
  - DWS 3.0 Decoupled Storage and Compute Usage Suggestions and Performance Optimization
- Data Development
- Database Management
- Performance Tuning
  - Optimizing Table Structure Design to Enhance GaussDB(DWS) Query Performance
  - Analyzing SQL Statements That Are Being Executed to Handle GaussDB(DWS) Performance Issues
- Cluster Management
  - Binding Different Resource Pools to Two Types of Jobs to Balance Load for GaussDB(DWS)
  - Scaling Options for GaussDB(DWS) with a Coupled Storage-Compute Architecture
Data Migration and Synchronization
- Data Migration to GaussDB(DWS)
- Importing Data
- Full Database Migration
  - Using CDM to Migrate Data to GaussDB(DWS)
  - Using DSC to Migrate SQL Scripts
- Real-time Import
  - Using DRS to Import Data to GaussDB(DWS)
  - Real-time Data Import From Kafka GaussDB(DWS)
- Metadata Migration
  - Using gs_dump and gs_dumpall to Export Metadata
  - Using gs_restore to Import Data
- Exporting Data
- Other Operations
Developer Guide
- Standard Data Warehouse (9.1.0.x)
- Standard data warehouse (8.1.3.x)
- Hybrid Data Warehouse (9.1.0.x）
- Hybrid Data Warehouse (8.1.3.x）
- Historical Versions
SQL Syntax Reference
- SQL Syntax Reference (9.1.0.x)
- SQL Syntax Reference (8.1.3.x)
- Historical Versions
Performance Tuning
- Overview of Query Performance Optimization
- Query Execution Process
- SQL Execution Plan
- SQL Optimization Guide
- Optimization Cases
- SQL Execution Troubleshooting
- query_band Load Identification
- Common Performance Parameter Optimization Design
Tool Guide
- Overview
- Downloading Related Tools
- gsql
- Data Studio
- GDS
- DSC
- DataCheck
- DWS-Connector
- Server Tool
API Reference
- Before You Start
- API Overview
- Calling APIs
- Getting Started
- API Description
- Application Cases
  - Using Postman to Call the API for Creating a Cluster
  - Using Postman to Call the API for Creating a Snapshot
- Introduction
- Appendix
SDK Reference
- SDK Overview
FAQs
- Top FAQs
- Product Consulting
- Database Connections
- Data Migration
- Database Usage
- Cluster Management
- Account Permissions
- Database Performance
- Backup and Restoration
  1. Why Does It Take a Long Time to Create an Automated Snapshot in GaussDB(DWS)?
  2. Does a GaussDB(DWS) Snapshot Have the Same Function as an EVS Snapshot?
Troubleshooting
- Database Connections
- JDBC/ODBC
- Data Import and Export
- Database Parameter Modification
- Account/Permission/Password
- Cluster Performance
- Cluster Exceptions
  - The Disk Usage Alarm Is Frequently Generated
- Database Use
Videos
Performance White Paper
- Overview
- Test Result
  - TPC-H Single Query Test
  - TPC-DS Single Query Test
- Test Methods
- Appendixes
  - TPC-H Test Sets
  - TPC-DS Test Sets
Technical White Paper
- GaussDB(DWS)
- Platforms and Technical Specifications Supported by GaussDB(DWS)
  - Technical Specifications
- GaussDB(DWS) Core Technologies
- GaussDB(DWS) Tools
  - Client Tools
  - Database Monitoring Tool
- External APIs
Error Code Reference
- 8.2.0 and earlier versions
  - Management Console Error Code
  - Data Warehouse Service Error Codes
- 8.2.1 or later versions
Glossary
More Documents
- User Guide
- API Reference (ME-Abu Dhabi Region)
- Developer Guide (ME-Abu Dhabi Region)
- SQL Syntax Reference (ME-Abu Dhabi Region)
- Tool Guide (ME-Abu Dhabi Region)
- Error Code Reference (ME-Abu Dhabi Region)
  - Management Console Error Code
- User Guide (Paris Region)
- API Reference (Paris Region)
- Developer Guide (Paris Region)
- SQL Syntax Reference (Paris Region)
- Tool Guide (Paris Region)
- Error Code Reference (Paris Region)
  - Management Console Error Code
- User Guide (Kuala Lumpur Region)
- API Reference (Kuala Lumpur Region)
- Developer Guide (Kuala Lumpur Region)
- SQL Syntax Reference (Kuala Lumpur Region)
- Tool Guide (Kuala Lumpur Region)
- Error Code Reference (Kuala Lumpur Region)
  - Management Console Error Code
General Reference
- Glossary
- Service Level Agreement
- White Papers
- Endpoints
- Permissions

On this page

Preparations
Step 1: Preparing MySQL Data
Step 2: Creating a GaussDB(DWS) Cluster
Step 3: Creating a DLI Queue
Step 4: Creating an Enhanced Datasource Connection
Step 5: Creating a DLI Flink Job
Step 6: Verifying Data Synchronization
More Information

Show all

Help Center/ GaussDB(DWS)/ Best Practices/ Data Migration/ Using a Flink Job of DLI to Synchronize MySQL Data to a GaussDB(DWS) Cluster in Real Time

Using a Flink Job of DLI to Synchronize MySQL Data to a GaussDB(DWS) Cluster in Real Time

Updated on 2024-10-29 GMT+08:00

View PDF

This practice demonstrates how to use a Flink job of DLI to synchronize MySQL data to GaussDB(DWS) in real time.

For details, see What Is Data Lake Insight?

This exercise lasts for approximately 60 minutes and involves utilizing various cloud services such as Virtual Private Cloud (VPC) and Subnet, Relational Database Service (RDS), Data Lake Insight (DLI), Object Storage Service (OBS), and GaussDB(DWS). The following is an outline of the exercise.

Preparations
Step 1: Preparing MySQL Data
Step 2: Creating a GaussDB(DWS) Cluster
Step 3: Creating a DLI Queue
Step 4: Creating an Enhanced Datasource Connection
Step 5: Creating a DLI Flink Job
Step 6: Verifying Data Synchronization
More Information

Preparations

You have registered a Huawei ID and enabled Huawei Cloud services.. The account cannot be in arrears or frozen.
You have created a VPC and subnet. For details, see Creating a VPC.

Step 1: Preparing MySQL Data

Buy an RDS instance and set the parameters listed in Table 1 (retain the default values for other parameters). For details, see Relational Database Service.

**Table 1** RDS parameters
Parameter	Value
Billing Mode	Pay-per-use
Region	CN-Hong Kong
DB Instance Name	rds-demo
DB Engine	MySQL
DB Engine Version	5.7 or later
Database Port	3306

Connect to the RDS instance and create an instance named mys_data.
1

CREATE DATABASE mys_data;

Switch to the new database mys_data and run the following command to create the mys_orders table:

      
         CREATE TABLE mys_data.mys_order
     ( order_id      VARCHAR(12),
       order_channel VARCHAR(32),
       order_time    DATETIME,
       cust_code     VARCHAR(6),
       pay_amount    DOUBLE,
       real_pay      DOUBLE,
       PRIMARY KEY (order_id) );

insert data to the table.

      
         INSERT INTO mys_data.mys_order VALUES ('202306270001', 'webShop', TIMESTAMP('2023-06-27 10:00:00'), 'CUST1', 1000, 1000);
INSERT INTO mys_data.mys_order VALUES ('202306270002', 'webShop', TIMESTAMP('2023-06-27 11:00:00'), 'CUST2', 5000, 5000);

Check whether the data is inserted.
1

SELECT * FROM mys_data.mys_order;

Step 2: Creating a GaussDB(DWS) Cluster

Creating a Cluster. To ensure network connectivity, select the same region and VPC as those of the RDS instance. In this practice, select China-Hong Kong. The VPC must be the same as that created for RDS.
Log in to the GaussDB(DWS) console, choose Dedicated Clusters > Clusters, locate the row that contains the target cluster, and click Login in the Operation column. The login information is as follows:
- Cluster: the created GaussDB(DWS) cluster.
- Database: gaussdb
- Data source name: dws-demo-01
- Username: dbadmin
- Password: password set when the GaussDB(DWS) cluster is created
Select Remember Password, click Test Connection, and wait until the connection is successful.
Copy the following SQL statements. In the SQL window, click Execute SQL to create a schema named dws_data.
1

CREATE SCHEMA dws_data;

Create the dws_order table in the new schema.

      
         CREATE TABLE dws_data.dws_order
        ( order_id      VARCHAR(12),
          order_channel VARCHAR(32),
          order_time    TIMESTAMP,
          cust_code     VARCHAR(6),
          pay_amount    DOUBLE PRECISION,
          real_pay      DOUBLE PRECISION );

Query data. The current table is empty.
1

SELECT * FROM dws_data.dws_order;

Step 3: Creating a DLI Queue

Log in to the Huawei Cloud console and choose Analytics > Data Lake Insight from the service list. The DLI console is displayed.
In the navigation pane on the left, choose Resources > Resource Pool.

Click Buy Resource Pool in the upper right corner, set the following parameters, and retain the default values for other parameters that are not described in the table.

**Table 2** DLI elastic resource pool
Parameter	Value
Billing Mode	Pay-per-use
Region	CN-Hong Kong
Name	dli_dws
Specifications	Standard
CIDR Block	172.16.0.0/18, which must be in a different network segment from MySQL and GaussDB(DWS). For example, if MySQL and GaussDB(DWS) are in the 192.168.x.x network segment, select 172.16.x.x for DLI.

Click Buy and click Submit.

After the resource pool is created, go to the next step.

On the elastic resource pool page, locate the row that contains the created resource pool, click Add Queue in the Operation column, and set the following parameters. Retain the default values for other parameters that are not described in the table.

**Table 3** Adding a queue
Parameter	Value
Name	dli_dws
Type	General purpose queue

Click Next and click OK. The queue is created.

Step 4: Creating an Enhanced Datasource Connection

In the security group of RDS, allow the network segment where the DLI queue is located.
1. In the navigation pane on the left, choose Resources > Queue Management and record the network segment of dli_dws.
  Figure 1 DLI queue network segment
2. Go to the RDS console, choose Instance Management in the navigation pane, and click the name of the created RDS instance.
3. Record the value of Private IP Address in the Connection Information area, which will be used in the subsequent connectivity test.
4. Click Manage next to the security group in Connection Information.
  Figure 2 RDS security group
5. In the security group list that is displayed, click the security group name to go to the security group configuration page.
6. Choose Inbound Rules > Add Rule, as shown in the following figure. Add the network segment of the DLI queue. In this example, the network segment is 172.16.0.0/18. Ensure that the network segment is the same as that entered in Step 3: Creating a DLI Queue.
  Figure 3 Adding a rule to the RDS security group
7. Click OK.
Return to the DLI console, click Datasource Connections on the left, select Enhanced, and click Create.

Set the following parameters. Retain the default values for other parameters that are not described in the table.

**Table 4** Connection from DLI to RDS
Parameter	Value
Connection Name	dli_rds
Resource Pool	Select the created DLI elastic resource pool.
VPC	Select the VPC where RDS is located.
Subnet	Select the subnet where RDS is located.
Other parameters	Retain the default values.

Figure 4 Creating a datasource connection

Click OK. Wait until the RDS connection is created.
Test the connectivity between DLI and RDS.
1. Choose Resources > Queue Management on the left, and choose More > Test Address Connectivity on the right of dli_dws.
2. Enter the private IP address of RDS recorded in 1.c and port 3306 in the address box.
3. Click Test to verify that DLI is successfully connected to RDS.
  Figure 5 Testing the connection between RDS and DLI
Test the connectivity between DLI and GaussDB(DWS).
1. Log in to the GaussDB(DWS) console, choose Dedicated Clusters > Clusters on the left, and click the cluster name to go to the details page.
2. As shown in the following figure, record the private IP address and port number of the GaussDB(DWS) cluster for future use.
  Figure 6 GaussDB(DWS) internal IP address
3. Click the security group name.
  Figure 7 GaussDB(DWS) security group
4. Choose Inbound Rules > Add Rule, as shown in the following figure. Add the network segment of the DLI queue. In this example, the network segment is 172.16.0.0/18. Ensure that the network segment is the same as that entered in 4.
  
  Figure 8 Adding a rule to the GaussDB(DWS) security group
5. Click OK.
6. Switch to the DLI console, choose Resources > Queue Management on the left, and click More > Test Address Connectivity on the right of dli_dws.
7. In the address box, enter the private IP address and port number of the GaussDB(DWS) cluster.
8. Click Test to verify that DLI is successfully connected to GaussDB(DWS).
  Figure 9 Testing GaussDB(DWS) connectivity

Step 5: Creating a DLI Flink Job

Log in to the OBS console and create an OBS bucket to store Flink jobs. For details, see the OBS User Guide.

Set key parameters as follows and retain the default values for other parameters.
- Region: CN-Hong Kong
- Bucket Name: dli-obs01 (If a conflict occurs, the bucket name can be increased from 02 to 03.)
- Bucket Policy: Private
Return to the DLI console, choose Job Management > Flink Jobs on the left, and click Create Job in the upper right corner.
Set Type to Flink OpenSource SQL and Name to rds-dws.

Figure 10 Creating a job
Click OK. The page for editing the job is displayed.
Set the following key parameters on the right of the page. Retain the default values for other parameters that are not described.
- Queue: Select dli_dws obtained in 4.
- Flink Version: Select version 1.15 or later. (The actual version is subject to the GUI.)
- OBS Bucket: Select the bucket created in 1 and click Authorize.
- (Optional) Select Save Job Log.

Copy the following SQL code to the SQL code window on the left.

For how to obtain the internal IP address of the RDS database, see 1.c. For details about how to obtain the internal IP address of the GaussDB(DWS) cluster, see 6.b. Change the password of user root of the RDS database and the password of user dbadmin of GaussDB(DWS).

       
        
          
          CREATE TABLE
  mys_order (
    order_id STRING,
    order_channel STRING,
    order_time TIMESTAMP,
    cust_code STRING,
    pay_amount DOUBLE,
    real_pay DOUBLE,
    PRIMARY KEY (order_id) NOT ENFORCED
  )
WITH
  (
    'connector' = 'mysql-cdc',
    'hostname' = 'Private IP address of the RDS DB instance',
    'port' = '3306',
    'username' = 'root',
    'password' = 'Password of user root of the RDS DB instance',
    'database-name' = 'mys_data',
    'table-name' = 'mys_order'
  );

CREATE TABLE
  dws_order (
    order_id STRING,
    order_channel STRING,
    order_time TIMESTAMP,
    cust_code STRING,
    pay_amount DOUBLE,
    real_pay DOUBLE,
    PRIMARY KEY (order_id) NOT ENFORCED
  )
WITH
  (
    'connector' = 'gaussdb',
    'driver' = 'com.huawei.gauss200.jdbc.Driver',
    'url' = 'jdbc:gaussdb://GaussDB(DWS) cluster private IP address:8000/gaussdb',
    'table-name' = 'dws_data.dws_order',
    'username' = 'dbadmin',
    'password' = 'Password of GaussDB(DWS) user dbadmin',
    'write.mode' = 'insert'
  );

INSERT INTO
  dws_order
SELECT
  *
FROM
  mys_order;

         

       
      

Click Format and click Save.

NOTICE:

Click Format to format the SQL code. Otherwise, new null characters may be introduced during code copy and paste, causing job execution failures.

Figure 11 Flink job parameters

Return to the DLI console home page and choose Job Management > Flink Jobs on the left.
Click Start on the right of the job name rds-dws and click Start Now.

Wait for about 1 minute and refresh the page. If the status is Running, the job is executed.

Figure 12 Running succeeded

Step 6: Verifying Data Synchronization

Go back to the SQL window of the GaussDB(DWS) database. If the connection times out, perform the following operations to log in again:
1. Go to the GaussDB(DWS) console.
2. In the navigation pane on the left, choose Dedicated Clusters > Clusters, and click Log In on the right of dws-demo.
Check whether two rows of data in the MySQL table have been synchronized to GaussDB(DWS).
1

SELECT * FROM dws_data.dws_order;
Figure 13 Query result

Switch to the RDS for MySQL page and run the following statements to insert three new data records:

      
         INSERT INTO mys_data.mys_order VALUES ('202403090003', 'webShop', TIMESTAMP('2024-03-09 13:00:00'), 'CUST1', 2000, 2000);
INSERT INTO mys_data.mys_order VALUES ('202403090004', 'webShop', TIMESTAMP('2024-03-09 14:00:00'), 'CUST2', 3000, 3000);
INSERT INTO mys_data.mys_order VALUES ('202403100004', 'webShop', TIMESTAMP('2024-03-10 10:00:00'), 'CUST3', 6000, 6000);

Figure 14 New MySQL data

Go back to the SQL window of GaussDB(DWS) and run the following SQL statement again. The returned result shows that the MySQL data has been synchronized to GaussDB(DWS) in real time.
1

SELECT * FROM dws_data.dws_order;
Figure 15 Real-time data synchronization

More Information

Storing authentication information for a data source directly in the job script for Flink cross-source development can result in password exposure. To enhance security, use DLI's datasource authentication function instead of specifying MySQL and GaussDB(DWS) usernames and passwords directly in job scripts.

NOTE:

Currently, only Flink 1.12 supports this function. Pay attention to the document changes on the official website.

Log in to the DLI console, click Datasource Connections, and click Datasource Authentication.
Click Create.
Create the password authentication for the root user of the MySQL database.
1. Set the following parameters:
  - Type: Password
  - Authentication Certificate: mysql_pwd_auth
  - Username: root
  - Password: password of user root
  Figure 16 MySQL password authentication
2. Click OK.
Create password authentication for the dbadmin user of GaussDB(DWS).
1. Set the following parameters:
  - Type: Password
  - Authentication Certificate: dws_pwd_auth
  - Username: dbadmin
  - Password: password of user dbadmin
  Figure 17 GaussDB(DWS) password authentication
2. Click OK.
On the DLI console, choose Job Management > Flink Jobs. Locate the row that contains the job created in Step 5: Creating a DLI Flink Job, and choose More > Stop to stop the job.
After the job is stopped, you can edit the job name.

Replace the SQL script with the latest one.

Replace the private IP addresses of RDS and GaussDB(DWS).

      
       
         
         CREATE TABLE mys_order (
  order_id STRING,
  order_channel STRING,
  order_time TIMESTAMP,
  cust_code STRING,
  pay_amount DOUBLE,
  real_pay DOUBLE,
  PRIMARY KEY (order_id) NOT ENFORCED ) 
WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'Private IP address of RDS',
  'port' = '3306',
  'pwd_auth_name' = 'mysql_pwd_auth',
  'database-name' = 'mys_data',
  'table-name' = 'mys_order' );
 
CREATE TABLE dws_order (
    order_id STRING,
    order_channel STRING,
    order_time TIMESTAMP,
    cust_code STRING,
    pay_amount DOUBLE,
    real_pay DOUBLE, 
    PRIMARY KEY (order_id) NOT ENFORCED )
WITH (
    'connector' = 'gaussdb',
    'driver' = 'com.huawei.gauss200.jdbc.Driver',
    'url' = 'jdbc:gaussdb://GaussDB(DWS) private IP address:8000/gaussdb',
    'table-name' = 'dws_data.dws_order',
    'pwd_auth_name' = 'dws_pwd_auth',
    'write.mode' = 'insert' );
 
INSERT INTO dws_order SELECT * FROM mys_order;

        

      
     

Click Format and click Save.
Restart the job and verify data synchronization by referring to Step 6: Verifying Data Synchronization.

Parent topic: Data Migration

Previous topic: Using a Flink Job of DLI to Synchronize Kafka Data to a GaussDB(DWS) Cluster in Real Time

Next topic: Migrating Data Between GaussDB(DWS) Clusters Using GDS

Feedback

Was this page helpful?

Helpful Not helpful

Provide feedback

Thank you very much for your feedback. We will continue working to improve the documentation.See the reply and handling status in My Cloud VOC.

The system is busy. Please try again later.

Which of the following issues have you encountered?

Content is inconsistent with the product UI

Unclear descriptions

Lack of examples or code

Incorrect steps

Can't find what I need

Lack of best practices

Feedback (optional)

0/500

Select at least one type of issue, and enter your comments or suggestions.

Enter a maximum of 500 characters.

Submit Cancel

For any further questions, feel free to contact us through the chatbot.

Chatbot