Migrate data from PostgreSQL to Kafka-Data Transmission Service(DTS)-阿里云帮助中心

Choose a migration type

Before you configure the task, decide which migration type fits your use case.

Migration type	When to use
Full data migration only	Migrate a snapshot of existing data. Do not write new data to the source during migration.
Full + incremental data migration	Zero-downtime migration. DTS first migrates existing data, then continuously replicates changes (INSERT, UPDATE, DELETE) from the source. Use this when you cannot take the source offline or need a record of deletions.

Schema migration is not supported when the destination Kafka instance uses Alibaba Cloud Instance as the access method.

Prerequisites

Before you begin, ensure that you have:

A destination Kafka instance with enough storage for the data to be migrated
A Kafka topic created in the destination Kafka instance to receive data
A privileged database account that is the owner of the database to be migrated. See Create an account
For incremental data migration: the wal_level parameter of the source RDS PostgreSQL instance set to logical. See Set instance parameters

For supported source and destination versions, see Migration solutions.

Billing

Migration type	Link configuration fees	Data transfer cost
Schema migration and full data migration	Free	Charged when Access Method is set to Public IP Address for the destination. See Billing overview.
Incremental data migration	Charged. See Billing overview.

Limitations

Source database

Bandwidth: The source database server must have outbound bandwidth of 100 Mb/s or more. Lower bandwidth reduces migration speed.
Object requirements:
- Tables must have primary keys or UNIQUE constraints with unique field values. Without these, duplicate records may appear in the destination.
- If the destination table was not created by DTS (that is, you did not select Schema Migration), the table must have the same primary key or a non-empty UNIQUE constraint as the source table.
- Database names cannot contain hyphens (for example, dts-testdata is not supported).
- DTS does not support migration of TimescaleDB extension tables, tables with cross-schema inheritance, or tables with expression-based unique indexes.
- For table-level migration with column mapping or other edits, a single task supports up to 1,000 tables. To migrate more tables, split them across multiple tasks or migrate the entire database in one task.
Write-ahead log (WAL) requirements for incremental migration:
- Set wal_level to logical.
- For incremental-only tasks: retain WAL logs for more than 24 hours.
- For tasks that include both full and incremental migration: retain WAL logs for at least 7 days. You can reduce the retention period to more than 24 hours after the full migration completes.
Important
If WAL logs are cleared before DTS reads them, the task fails and may cause data inconsistency or data loss. Issues caused by insufficient log retention are not covered by the DTS Service-Level Agreement (SLA).
DDL operation limits:
- Non-standard DDL syntax may cause the migration task to fail or lose data.
- Performing a major engine version upgrade on the source database while a DTS instance is running causes the instance to fail permanently. Reconfigure the DTS instance after the upgrade.
- During full migration, do not perform DDL operations that change the database or table schema.
- If a single incremental data record exceeds 256 MB, the DTS instance fails permanently. Reconfigure the instance.
- During full-only migration, do not write new data to the source. For real-time consistency, select both Full Data Migration and Incremental Data Migration.
Logical replication slot failover: To prevent logical replication interruptions caused by RDS failover, enable Logical Replication Slot Failover on the source RDS PostgreSQL instance. See Logical Replication Slot Failover.
Long-running transactions: If the source database has long-running transactions during incremental migration, WAL logs before the transaction commit cannot be cleared. This causes log accumulation and may exhaust disk space.

Other limitations

A single migration task covers one database. To migrate multiple databases, create a separate task for each.
REPLICA IDENTITY requirement for incremental migration: Before writing data to the source tables, run the following command on each table to be migrated:
- When the instance runs for the first time.
- When migration granularity is Schema, and a new table is created or an existing table is rebuilt using RENAME.
```
ALTER TABLE schema.table REPLICA IDENTITY FULL;
```
Replace schema and table with the actual schema and table names. Run this command during off-peak hours to avoid table locking. DTS requires this command in two situations: If you skip the related precheck item, DTS automatically runs this command during instance initialization.
Temporary tables created by DTS: DTS creates the following temporary tables in the source database to capture DDL statements, table structures, and heartbeat information. Do not delete them during migration — they are removed automatically when the DTS instance is released. public.dts_pg_class, public.dts_pg_attribute, public.dts_pg_type, public.dts_pg_enum, public.dts_postgres_heartbeat, public.dts_ddl_command, public.dts_args_session, public.aliyun_dts_instance
Replication slot management: During incremental migration, DTS creates a replication slot with the prefix dts_sync_ in the source database. This slot retains WAL segments and allows DTS to retrieve incremental logs from the last 15 minutes. When migration fails or the instance is released, DTS attempts to clear the slot automatically.

Warning
If you change the password of the source database account or remove the DTS IP addresses from the source database whitelist during migration, the replication slot cannot be cleared automatically. Manually clear the slot in the source database to prevent disk space exhaustion. If a failover occurs, log on to the secondary database to clear the slot.
Full migration resource usage: DTS consumes read and write resources on both the source and destination during full migration, which increases database load. Run migrations during off-peak hours (for example, when CPU load is below 30%). Full migration uses concurrent INSERT operations, so destination tables may occupy more space than source tables due to fragmentation.
Running instance behavior:
- If the destination Kafka instance is scaled out or in while the migration instance is running, restart the migration instance.
- If data is written to the destination from a source other than DTS, data inconsistency may occur and the migration instance may fail.
Metadata validation: DTS validates data content but does not validate metadata such as Sequences. Validate this metadata separately.
Task recovery: If a task fails, DTS support attempts recovery within 8 hours. Recovery may involve restarting the task or adjusting parameters. Only DTS task parameters are modified — database parameters remain unchanged.
Partitioned tables: Include both the parent table and all child partitions as migration objects. The parent table does not store data directly — all data is in the child partitions. Omitting any partition causes data inconsistency.

Special cases

For RDS PostgreSQL source instances: do not change the endpoint or zone during migration. This causes the migration to fail.
For self-managed PostgreSQL source databases: ensure max_wal_senders and max_replication_slots are each greater than the sum of replication slots currently in use and the number of DTS instances being created.
For Google Cloud SQL for PostgreSQL: use an account with the cloudsqlsuperuser permission. When selecting migration objects, choose only objects that this account owns, or grant it Owner permission using:

An account with cloudsqlsuperuser permission cannot manage objects owned by another cloudsqlsuperuser account.
```
GRANT <owner_of_objects_to_migrate> TO <source_db_account_for_task>;
```

SQL operations supported for incremental migration

Operation type	SQL operations
DML	`INSERT`, `UPDATE`, `DELETE`
DDL	`CREATE TABLE`, `DROP TABLE`, `ALTER TABLE` (including `RENAME TABLE`, `ADD COLUMN`, `ADD COLUMN DEFAULT`, `ALTER COLUMN TYPE`, `DROP COLUMN`, `ADD CONSTRAINT`, `ADD CONSTRAINT CHECK`, `ALTER COLUMN DROP DEFAULT`), `TRUNCATE TABLE` (PostgreSQL 11 or later only), `CREATE INDEX ON TABLE`

DDL migration requirements:

Only tasks created after May 12, 2023 support DDL migration. For tasks created before that date, create triggers and functions in the source database to capture DDL before configuring the task. See Use triggers and functions to implement incremental DDL migration for PostgreSQL.
BIT type data cannot be migrated during incremental migration.
The source database account must be a privileged account.
The minor engine version of the RDS PostgreSQL instance must be 20210228 or later. See Upgrade the minor engine version.

Unsupported DDL patterns:

DDL with CASCADE or RESTRICT modifiers
DDL in sessions using SET session_replication_role = replica
DDL executed by calling a FUNCTION
Transactions containing both DML and DDL statements
DDL for objects outside the migration scope
DDL executed via the Server Programming Interface (SPI) inside a plugin

Create a migration task

Step 1: Open the migration task list

Use one of the following methods to navigate to the migration task list for your target region.

From the DTS console

Log on to the DTS console.
In the left navigation pane, click Data Migration.
In the upper-left corner, select the region where the migration instance will be created.

From the DMS console

DMS console operations may vary based on your console mode and layout. See Simple mode console and Customize the layout and style of the DMS console.

Log on to the Data Management (DMS) console.
In the top menu bar, choose Data + AI > Data Transmission (DTS) > Data Migration.
To the right of Data Migration Tasks, select the region where the migration instance will be created.

Step 2: Configure source and destination databases

Click Create Task.

Configure the task and source database.

To reuse a previously registered database connection, select it from the Select Existing Connection drop-down list. In the DMS console, this parameter is named Select a DMS database instance.

Parameter	Value
Task Name	Enter a descriptive name for easy identification. Names do not have to be unique.
Database Type (source)	Select PostgreSQL.
Access Method (source)	Select Alibaba Cloud Instance.
Instance Region	Select the region of the source ApsaraDB RDS for PostgreSQL instance.
Replicate Data Across Alibaba Cloud Accounts	Select No if the source is in the same Alibaba Cloud account.
Instance ID	Select the ID of the source RDS PostgreSQL instance.
Database Name	Enter the name of the database containing the objects to migrate.
Database Account	Enter a privileged account that owns the database.
Database Password	Enter the password for the database account.

Configure the destination database.

Parameter	Value
Database Type (destination)	Select Kafka.
Access Method (destination)	Select Alibaba Cloud Instance.
Instance Region	Select the region of the destination Kafka instance.
Replicate Data Across Alibaba Cloud Accounts	Select No if the destination is in the same Alibaba Cloud account.
Kafka Instance ID	Select the ID of the destination Kafka instance.
Encryption	Select a method based on your Kafka instance settings and security requirements.
Topic	Select the topic to receive migrated data.
Use Kafka Schema Registry	Select Yes to use Kafka Schema Registry and enter the Avro schema URL or IP address. Select No to skip. Kafka Schema Registry is a metadata service providing a RESTful interface for storing and retrieving Avro schemas.

Click Test Connectivity and Proceed.

Add the DTS server IP address ranges to the security settings (whitelist) of both the source and destination databases. See Add DTS server IP addresses to a whitelist. For self-managed databases (where Access Method is not Alibaba Cloud Instance), also click Test Connectivity in the CIDR Blocks of DTS Servers dialog.

Step 3: Configure migration objects

On the Configure Objects page, set the following parameters.

If you use object name mapping, migrations of other objects that depend on the mapped object may fail.

Parameter	Description
Migration Types	Select Full Data Migration for a one-time snapshot. Select both Full Data Migration and Incremental Data Migration for zero-downtime migration. Do not write new data to the source during full-only migration.
Processing Mode of Conflicting Tables	Precheck and Report Errors: Reports an error if tables with the same name exist in the destination. Ignore Errors and Proceed: Skips the check and continues. Use with caution — this may cause data inconsistency if schemas differ or if the same primary key exists in both source and destination.
Data Format in Kafka	Select the storage format for migrated data: Canal JSON (see Canal JSON format), DTS Avro (parse using the DTS Avro schema and DTS Avro deserialization example), or Shareplex JSON (see Shareplex JSON).
Kafka data compression format	LZ4 (default): low compression ratio, high speed. GZIP: high ratio, low speed, high CPU usage. Snappy: medium ratio, medium speed.
Policy for shipping data to Kafka partitions	Select a partition shipping policy.
Message acknowledgement mechanism	Select a message acknowledgement mechanism.
Topic that stores DDL information	Select the topic for DDL information. If none is selected, DDL information is stored in the data-receiving topic.
Capitalization of object names in destination instance	Configure case sensitivity for migrated object names. Default is DTS default policy. See Case sensitivity of object names.
Source Objects	In the Source Objects box, select objects at the schema or table level, then click to move them to Selected Objects.
Selected Objects	To set topic name, partition count, and partition key for a source table, hover over the table and click Edit. See Map migration objects to Kafka topics. To filter SQL operations for incremental migration, right-click the object and select the desired operations.

Click Next: Advanced Settings and configure advanced parameters.

Parameter	Description
Dedicated cluster for task scheduling	DTS uses a shared cluster by default. For more stable performance, purchase a dedicated cluster.
Retry time for failed connections	Default is 720 minutes (range: 10–1440 minutes). Set to at least 30 minutes. DTS resumes the task if it reconnects within this period; otherwise the task fails. DTS charges during the retry period.
Retry time for other issues	Default is 10 minutes (range: 1–1440 minutes). Set to at least 10 minutes. Must be less than Retry time for failed connections.
Enable throttling for full data migration	Set QPS (queries per second) to the source, RPS (records per second) of full migration, and data migration speed (MB/s) to reduce source and destination load. Available only when Full Data Migration is selected. You can also adjust the speed after the task starts.
Enable throttling for incremental data migration	Set RPS and data migration speed (MB/s) for incremental migration. Available only when Incremental Data Migration is selected.
Timeout for modifying table REPLICA IDENTITY to FULL during initialization	Maximum wait time for a single `ALTER TABLE ... REPLICA IDENTITY FULL` DDL during task initialization.
Retry interval for modifying the table's REPLICA IDENTITY to FULL during initialization	Interval between retries when the REPLICA IDENTITY DDL times out.
Environment Tag	(Optional) Tag the instance with an environment label.
Scene Label	(Optional) Tag the instance with a scenario label. Does not affect task operation.
Configure ETL	Select Yes to use the ETL feature and enter data processing statements. Select No to skip.
Monitoring and alerting	Select Yes to configure an alert threshold and notification. DTS sends alerts if migration fails or latency exceeds the threshold.

Step 4: Run a precheck

Click Next: Save Task Settings and Precheck.

To view the OpenAPI parameters for this configuration, hover over the Next: Save Task Settings and Precheck button and click Preview OpenAPI parameters.
DTS performs a precheck before starting the task. If the precheck fails:
- Click View Details next to the failed item, fix the issue, and rerun the precheck.
- For warnings on non-ignorable items: fix and rerun.
- For warnings on ignorable items: click Confirm Alert Details > Ignore > OK > Precheck Again. Ignoring warnings may cause data inconsistency.

Step 5: Purchase and start the instance

When the precheck Success Rate reaches 100%, click Next: Purchase Instance.

On the Purchase page, select the instance class.

Parameter	Description
Resource Group Settings	Select the resource group for the instance. See What is Resource Management?
Instance Class	Select a specification based on required migration speed. See Data migration link specifications.

Read and accept Data Transmission Service (Pay-as-you-go) Service Terms.
Click Buy and Start, then click OK in the confirmation dialog.

Verify migration status

View task progress on the Data Migration Tasks list page.

Full migration only: The task stops automatically after full migration completes. The Status changes to Completed.
Full + incremental migration: The task continues running after full migration completes. The Status shows Running while incremental migration is active.

Map migration objects to Kafka topics

Use the mapping feature to set the topic name, partition count, and partition key for each source table in the destination Kafka instance.

In the Selected Objects area, hover over the destination topic name at the table level.
Click Edit next to the topic name.

In the Edit Table dialog, configure the mapping.

At the schema level, the dialog is named Edit Schema and supports fewer parameters. If you are not migrating an entire schema, Name of target topic and Number of partitions cannot be changed in the Edit Schema dialog.

Parameter	Description
Name of target topic	The destination topic for this source table. Defaults to the topic selected in the destination database configuration. For Alibaba Cloud Kafka instances, the topic must already exist. If it does not, migration fails. Changing this field routes data to the specified topic.
Filter conditions	See Set filter conditions.
Number of partitions	The number of partitions for the destination topic.
Partition key	Available when Policy for Shipping Data to Kafka Partitions is set to Ship Data to Separate Partitions Based on Hash Values of Primary Keys. Specify one or more columns as the partition key. DTS delivers rows to different partitions based on the hash value of the key columns. Only configurable in the Edit Table dialog (not the Edit Schema dialog).

Click OK.