Simplify MySQL to PostgreSQL Migration

This article provides a detailed guide on how to achieve full data synchronization from MySQL to PostgreSQL using Apache SeaTunnel 2.3.9. It covers the entire process, from environment setup to production validation.

Version Requirements:

• MySQL: MySQL 8.3
• PostgreSQL: PostgreSQL 13.2
• Apache SeaTunnel: Apache-SeaTunnel-2.3.9

Version Verification:

Run the following SQL command to verify the version:

  
select version();  

Enabling Master-Slave Replication:

  
show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');  

For MySQL CDC data synchronization, SeaTunnel needs to read the binlog and act as a slave node in the cluster.

  
SET GLOBAL gtid_mode=OFF_PERMISSIVE;  
SET GLOBAL gtid_mode=ON_PERMISSIVE;  
SET GLOBAL enforce_gtid_consistency=ON;  
SET GLOBAL gtid_mode=ON;  

Granting User Permissions:

A user with replication privileges is required:

  
CREATE USER 'test'@'%' IDENTIFIED BY 'password';  
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'test';  
FLUSH PRIVILEGES;  

SeaTunnel Cluster Configuration:

By default, SeaTunnel generates logs in a single file. For production, it is recommended to separate logs by task:

  
rootLogger.appenderRef.file.ref = routingAppender  

In production environments, it is recommended to install SeaTunnel in the /opt directory and configure the SEATUNNEL_HOME environment variable:

  
ln -s /opt/apache-seatunnel-2.3.9 /opt/seatunnel  
export SEATUNNEL_HOME=/opt/seatunnel  

SeaTunnel Job Configuration:

  
env {  
  job.mode = 'STREAMING'  
  job.name = 'DEMO'  
  parallelism = 3  
  checkpoint.interval = 30000  
  checkpoint.timeout = 30000  
  job.retry.times = 3  
  job.retry.interval.seconds = 3  
}  

Setting a job name is essential for managing tasks in production. Additionally, a parallelism level of 3 is set, but it can be adjusted according to the cluster size.

Data Source Configuration (MySQL CDC):

  
source {  
  MySQL-CDC {  
    base-url = 'jdbc:mysql://192.168.8.101:3306/test?serverTimezone=Asia/Shanghai'  
    username = 'test'  
    password = '123456'  
    database-names = ['test']  
    table-pattern = 'test\.test_.*'  
    startup.mode = 'initial'  
    snapshot.split.size = '8096'  
    snapshot.fetch.size = '1024'  
    server-id = '6500-8500'  
    exactly_once = false  
    schema-changes.enabled = true  
  }  
}  

Synchronization is configured for both historical and incremental data. Enabling schema evolution is recommended to handle changes in table structures.

Destination Configuration (PostgreSQL):

  
sink {  
  jdbc {  
    url = 'jdbc:postgresql://192.168.8.101:5432/test'  
    driver = 'org.postgresql.Driver'  
    user = 'postgres'  
    password = '123456'  
    generate_sink_sql = true  
    database = 'test'  
    table = '${database_name}.${table_name}'  
    schema_save_mode = 'CREATE_SCHEMA_WHEN_NOT_EXIST'  
    data_save_mode = 'APPEND_DATA'  
  }  
}  

The generate_sink_sql option automatically generates insert and update statements. The schema save mode is also configured to automatically create tables if they do not exist.

Task Submission and Monitoring:

  
./bin/start-seatunnel.sh --config /path/to/config.yaml --async  

The --async flag allows the task to run in the background while being monitored through the SeaTunnel interface.

Data Synchronization Demonstration:

Inserts, updates, and deletes were tested in MySQL, and the changes were successfully replicated in PostgreSQL.

About Apache SeaTunnel:

• Supports hundreds of data sources.
• Handles full, incremental, and real-time synchronization.
• Effective monitoring to prevent data loss or duplication.

At Q2BSTUDIO, a company specialized in development and technology services, we help implement data synchronization solutions to ensure data integrity and efficiency in enterprise environments. With experience in integrating Apache SeaTunnel, we provide consulting and development of customized strategies to optimize data transfer between various platforms, ensuring performance and scalability.