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

Show all

Help Center/ GaussDB(DWS)/ User Guide/ Migrating Service Data to a GaussDB(DWS) Cluster/ Migrating Data to a GaussDB(DWS) Cluster Using GDS-Kafka/ Managing Instances

Managing Instances

Updated on 2025-03-03 GMT+08:00

View PDF

Overview

Data migration provides independent clusters for secure and reliable data migration. Clusters are isolated from each other and cannot access each other. With instance management, you can easily create and manage clusters by purchasing GDS-Kafka instances. GDS-Kafka consumes and caches data from Kafka. If the data cache time or size reaches a preconfigured threshold, GDS-Kafka will copy the data to a GaussDB(DWS) temporary table, and then insert or update data in the temporary table.

The format of data generated by the Kafka message producer is specified by the kafka.source.event.type parameter. For details, see Message Formats Supported by GDS-Kafka.
In GDS-Kafka, you can directly insert data for tables without primary keys, or update data by merging. Direct insert can achieve better performance, because it does not involve update operations. Determine your update mode based on the target table type in GaussDB(DWS). The data import mode is determined by the app.insert.directly parameter and whether a primary key exists. For details, see GDS-Kafka Data Import Modes.

NOTE:

GDS-kafka only allows lowercase target table and column names.
GDS-Kafka deletes historical data based on pos in the extended field. If imported data involves the delete operation, the extended field must be used.

Purchasing a GDS-Kafka Instance

To use the data migration feature, you need to purchase a GDS-kafka instance (cluster). Cluster instances provide secure and reliable data migration services. Clusters are isolated from each other.

Constraints

Currently, only standalone clusters are supported.
Only the pay-per-use billing mode is supported.

Procedure

Log in to the GaussDB(DWS) console.
In the navigation pane, choose Data > Data Integration > Instances.

In the upper right corner of the page, click Buy GDS-Kafka Instance. Configure cluster parameters.

**Table 1** Parameter description
Parameter	Description	Example Value
CPU Architecture	The following CPU architectures can be selected: x86 Kunpeng NOTE: The x86 and Kunpeng architectures differ only in their underlying structure, which is not sensible to the application layer. Both architectures use the same SQL syntax. If you need to create a cluster and find that x86 servers are not enough, you can opt for the Kunpeng architecture.	x86
Flavor	Select a node flavor.	-
Capacity	Storage capacity of a node.	-
Current Flavor	Current flavor of the cluster.	-
Name	Set the name of the data warehouse cluster. Enter 4 to 64 characters. Only case-insensitive letters, digits, hyphens (-), and underscores (_) are allowed. The value must start with a letter. Letters are not case-sensitive.	-
Version	Version of the database instance installed in the cluster.	-
VPC	Specify a VPC to isolate the cluster's network. If you create a data warehouse cluster for the first time and have not configured the VPC, click View VPC. On the VPC management console that is displayed, create a VPC as needed.	-
Subnet	Specify a VPC subnet. A subnet provides dedicated network resources that are isolated from other networks, improving network security.	-
Security Group	Specify a VPC security group. A security group restricts access rules to enhance security when GaussDB(DWS) and other services access each other.	-
EIP	Specify whether users can use a client to connect to a cluster's database over the Internet. The following methods are supported: Do not use: Do not specify any EIPs here. If GaussDB(DWS) is used in the production environment, first bind it to ELB, and then bind it to an EIP on the ELB page. Buy now: Specify bandwidth for EIPs, and the system will automatically assign EIPs with dedicated bandwidth to clusters. You can use the EIPs to access the clusters over the Internet. The bandwidth name of an automatically assigned EIP starts with the cluster name. Specify: Specify an EIP to be bound to the cluster. If no available EIPs are displayed in the drop-down list, click Create EIP to go to the Elastic IP page and create an EIP as needed. The bandwidth can be customized.	-
Enterprise Project	Select the enterprise project of the cluster. You can configure this parameter only when the Enterprise Project Management service is enabled. The default value is default.	default

If the configuration is correct, click Buy Now.

Viewing Instance Details

On the instance details page, you can view the basic information and network information about the cluster.

Procedure

Log in to the GaussDB(DWS) console.
In the navigation pane, choose Data > Data Integration > Instances.
Click the name of an instance to go to the instance details page.

Figure 1 Viewing Instance Details

Message Formats Supported by GDS-Kafka

Table 2 Message formats supported by GDS-Kafka

Kafka Source Event Type

Format

Description

cdc.drs.avro

Internal format of Huawei Cloud DRS. DRS generates data in the avro format used by Kafka. GDS-Kafka can directly interconnect with DRS to parse and import the data.

None

drs.cdc

To use the avro format for drs.cdc, specify the Maven dependency of GDS-Kafka-common and GDS-Kafka-source in the upstream programs of Kafka, and then create and fill in the Record object. A Record object represents a table record. It will be serialized into a byte[] array, produced and sent to Kafka, and used by the downstream GDS-Kafka.

In the following example, the target table is the person table in the public schema. The person table consists of the id, name, and age fields. The op_type is U, which indicates an update operation. This example changes the name field from a to b in the record with the ID 0, and changes the value of the age field from 18 to 20.

         
          
            
            Record record = new Record();
// Set the schema and table name of the target table.
record.setTableName("public.person");
// Set the field list.
List<Field> fields = new ArrayList<>();
fields.add(new Field("id", 0));
fields.add(new Field("name", 1));
fields.add(new Field("age", 2));
record.setFields(fields);
// Set the field value list before the table record is updated.
List<Object> before = new ArrayList<>();
before.add(new Integer(0, "0"));
before.add(new Character("utf-8", ByteBuffer.wrap("a".getBytes(StandardCharsets.UTF_8))));
before.add(new Integer(0, "18"));
record.setBeforeImages(before);
// Set the field value list after the table record is updated.
List<Object> after = new ArrayList<>();
after.add(new Integer(0, "0"));
after.add(new Character("utf-8", ByteBuffer.wrap("b".getBytes(StandardCharsets.UTF_8))));
after.add(new Integer(0, "20"));
record.setAfterImages(after);
// Set the operation type.
record.setOperation("U");
// Set the operation time.
record.setUpdateTimestamp(325943905);
// Serialize the record object into a byte[] array.
byte[] msg = Record.getEncoder().encode(record).array();

           

         
        

Standard avro format:

The tableName field is used to describe the target table and schema names that the current record belongs to. [Mandatory]
The operation field is used to describe the operation type of the current record. I indicates insert, U indicates update, and D indicates deletion. [Mandatory]
updateTimestamp indicates the time when an operation is performed on the source end. [Optional]
The beforeImages list describes the information before the current record is updated or deleted. The fields in the before body correspond to those in the target table. [Mandatory for U/D]
The afterImages list describes the updated or newly inserted information of the current record. [Mandatory for U/D]
The fields list describes the field list of the current table record. The index values of the fields must be in the same sequence as those in beforeImage and afterImage. [Mandatory]

cdc.json

In the following example, the target table is the person table in the public schema. The person table consists of the id, name, and age fields. The op_type is U, which indicates an update operation. This example changes the name field from a to b in the record with the ID 1, and changes the value of the age field from 18 to 20.

         
          
            
            {
"table": "public.person",
"op_type": "U",
"op_ts": "1668426344",
"current_ts": "1668426344",
"before": {
"id":"1",
"name":"a",
"age": 18
},
"after": {
"id":"1",
"name":"b",
"age": 20
}
}

           

         
        

Standard JSON format:

The table field describes the target table and schema names that the current record belongs to. [Mandatory]
The op_type field is used to describe the operation type of the current record. I indicates insert, U indicates update, and D indicates deletion. [Mandatory]
op_ts indicates the time when an operation is performed on the source end. [Optional]
current_ts indicates the time when a message is imported to Kafka. [Optional]
The before object describes the information before the current record is updated or deleted. The fields in the before body correspond to those in the target table. [Mandatory for U/D]
The after object list describes the update or newly inserted information of the current record. [Mandatory for U/D]

industrial.iot.json

         
          
            
            {
"header": {
"thing_id":"a0001",
"instance_id":"1",
"thing_model_name":"computer",
"timestamp":"1668426344"
},
"body": {
"status":"Normal",
"temperature":"10",
"working_time":"10000"
},
}

           

         
        

IoT data format:

thing_model_name in header indicates the table name. [Mandatory]
The values of thing_id, instance_id, and timestamp in header and the content in the body comprise the fields of the current record.
IoT data is time series data and does not involve update or deletion. Only insert operations are involved.

industrial.iot.recursion.json

         
          
            
            {
"header": {
"thing_id":"a0001",
"instance_id":"1",
"thing_model_name":"computer",
"timestamp":"1668426344"
},
"body": {
"status":"Normal",
"temperature":"10",
"property":{
  "key1":"1",
  "key2":2
},
"working_time":"10000"
},
}

           

         
        

IoT data format:

thing_model_name in header indicates the table name. [Mandatory]
The values of thing_id, instance_id, and timestamp in header and the content in the body comprise the fields of the current record.
IoT data is time series data and does not involve update or deletion. Only insert operations are involved.
In this data format, the key and value of body are added to the property and value fields in the new format to generate multiple pieces of new data. In this way, rows are converted to columns.

industrial.iot.event.json.independent.table

         
            {
"event_id":"1",
"event_name":"test",
"start_time":"1970-1-1T00:00:00.000Z",
"end_time":"1970-1-1T00:00:00.000Z",
"fields":{
    "field1":"value1",
    "field2":2
    }
}

IoT event stream data format:

event_name indicates a table name. [Mandatory]
event_id, start_time, end_time, and fields comprise the field content of a record. [Mandatory]
IoT event stream data is time series data and does not involve update or deletion. Only insert operations are involved.

industrial.iot.json.multi.events

         
          
            
            {
"event_id":"1",
"event_name":"test",
"start_time":"1970-1-1T00:00:00.000Z",
"end_time":"1970-1-1T00:00:00.000Z",
"fields":{
    "field1":"value1",
    "field2":2,
    "field3":{
       "key1":"1",
       "key2":2
       }
    }
}

           

         
        

IoT event stream data format:

event_name indicates a table name. [Mandatory]
event_id, start_time, end_time, and fields comprise the field content of a record. [Mandatory]
IoT event stream data is time series data and does not involve update or deletion. Only insert operations are involved.

In this data format, the key and value of fields are added to the field_name and field_value fields in the new format to generate multiple pieces of new data. In this way, rows are converted to columns.

GDS-Kafka Import Modes

To import GDS-Kafka data to the database, copy the data to a temporary table, and then merge or insert the data. The following table describes their usage and scenarios.

**Table 3** GDS-Kafka import modes
Operation	Direct Insertion	Primary Key Table	Import Mode
insert	true (only for tables without primary keys)	No	Use INSERT SELECT to write data from the temporary table to the target table.
	false	Yes	Merge data from the temporary table to the target table based on the primary key.
	false	No	Use INSERT SELECT to write data from the temporary table to the target table.
delete	true (only for tables without primary keys)	No	Use INSERT SELECT to write data from the temporary table to the target table.
	false NOTE: You can mark deletion by configuring the app.del.flag parameter. The flag of a deleted record will be set to 1.	Yes	If the delflag field is set, merge will be performed based on the primary key. If a matched primary key is found, and the value of pos in the target table is smaller than that in the temporary table, the delflag field will be set to 1. Otherwise, a new record will be inserted. If the delflag field is not set, a matched primary key is found, and the value of pos in the target table is smaller than that in the temporary table, the record will be deleted from the target table.
		No	If the delflag field is set, all the fields in the temporary table will be used to match and merge with the target table. If a matched record is found, and the value of pos in the target table is smaller than that in the temporary table, the delflag field will be set to 1. Otherwise, a new record will be inserted. If the delflag field is not set, all the fields in the temporary table will be used to match the target table. If a matched record is found, and the value of pos in the target table is smaller than that in the temporary table, the matched record will be deleted from the target table.
update	true (only for tables without primary keys)	No	Use INSERT SELECT to write data from the temporary table to the target table.
	false NOTE: The update operation is split. The message in before or beforeImage is processed as a delete operation, and the message in after or afterImage is processed as an insert operation. Then, the message is saved to the database based on the insert and delete operations.	Yes	Equivalent to the insert+delete operation on a table with a primary key.
		No	Equivalent to the insert+delete operation on a table without a primary key.