Migrate Data from Transactional Databases

This document describes how to migrate data from traditional transactional (OLTP / TP) databases such as MySQL, SQL Server, Oracle, and PostgreSQL into your warehouse. VeloDB Cloud provides several data ingestion options that cover one-time migration, offline batch synchronization, and real-time incremental synchronization.

Before you begin

For supported MySQL and PostgreSQL sources, Data Integration provides a guided full and incremental import. Check its source prerequisites and deployment availability. The SQL and connector options below cover one-time migrations, custom processing, and existing pipelines. Native Streaming Jobs support MySQL and PostgreSQL; do not assume the same support for every transactional database.

Use a running warehouse and the connection values from Connect. Replace source hosts, driver URLs, and credentials in the examples. Use source accounts with the required read or CDC privileges and a warehouse SQL user with the required load and create privileges. Flink, Spark, and third-party tools run in your environment. Create the target databases first. Pre-create matching tables for INSERT INTO and connector writes, or use CTAS or auto-table mode where indicated. Validate snapshot counts, type and key mappings, subsequent updates and deletes, and CDC lag before switching workloads.

Overview of migration options

The following table summarizes the main migration options that VeloDB Cloud supports. You can choose one based on data volume, synchronization latency requirements, and whether CDC (Change Data Capture) is needed.

Option	Applicable scenario	Sync mode	Whole-database supported	Incremental / CDC supported
Multi-Catalog	One-time migration, ad hoc queries, CTAS table creation	Offline / on-demand	Implement with SQL	No
Flink Doris Connector (JDBC)	Offline batch synchronization in the Flink stack	Offline	No	No
Flink Doris Connector (CDC)	Real-time full + incremental synchronization	Real-time	Supported	Supported (INSERT / UPDATE / DELETE)
Spark Connector	Batch data writes in the Spark stack	Offline	No	No
Streaming Job	Built-in continuous synchronization from MySQL / PostgreSQL in your warehouse	Streaming (continuous)	Supported	Supported
Third-party tools (NineData / DataX / SeaTunnel / CloudCanal)	Existing synchronization platforms, visual operations	Tool and source dependent	Tool and source dependent	Tool and source dependent

Note:

If you need to complete the migration without introducing an external compute engine, use Multi-Catalog or Streaming Job. If you need real-time incremental synchronization, use Flink Doris Connector + CDC, Streaming Job, or NineData for MySQL-compatible sources to VeloDB Cloud.

---

Option 1: multi-catalog (recommended for one-time migration)

Use a JDBC Catalog to map tables in the TP database as external tables in your warehouse, then use INSERT INTO or CREATE TABLE AS SELECT (CTAS) to load the data.

Steps

1. Create a JDBC Catalog that points to the source TP database.
2. Use INSERT INTO to write into an existing VeloDB Cloud table, or use CTAS to create the table and load data in a single step.
3. Verify row counts and field consistency.

Example: migrate data from MySQL

migrating data from MySQL

-- INSERT and CTAS below are alternatives. Run only one for the same target table.
CREATE CATALOG mysql_catalog properties(
    'type' = 'jdbc',
    'user' = '<source-user>',
    'password' = '<password>',
    'jdbc_url' = 'jdbc:mysql://<mysql-host>:3306/mysql_db',
    'driver_url' = '<jdbc-driver-url>',
    'driver_class' = 'com.mysql.cj.jdbc.Driver'
);

-- Load into an existing table with INSERT INTO
INSERT INTO internal.doris_db.tbl1
SELECT * FROM mysql_catalog.mysql_db.table1;

-- Create the table and load data in one step with CTAS
CREATE TABLE internal.doris_db.tbl1
AS
SELECT * FROM mysql_catalog.mysql_db.table1;

For more details, see Catalog Data Import.

---

Option 2: Flink Doris Connector

With Flink, you can run TP-system synchronization in either offline batch mode or real-time incremental mode.

2.1 Offline synchronization: Flink JDBC source + Doris Sink

This option suits offline data movement in the Flink stack. The following Flink SQL example shows the pattern:

CREATE TABLE student_source (
    id INT,
    name STRING,
    age INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://<mysql-host>:3306/mydatabase',
  'table-name' = 'students',
  'username' = 'username',
  'password' = '<password>'
);

CREATE TABLE student_sink (
    id INT,
    name STRING,
    age INT
    )
    WITH (
      'connector' = 'doris',
      'fenodes' = '<warehouse-http-host>:<http-port>',
      'table.identifier' = 'test.students',
      'username' = '<warehouse-user>',
      'password' = '<password>',
      'sink.label-prefix' = 'doris_label'
);

INSERT into student_sink select * from student_source;

For more details, see Flink JDBC.

2.2 Real-time synchronization: Flink CDC

With Flink CDC, you can perform a full data read and incremental capture in a single job, and synchronize INSERT / UPDATE / DELETE events.

Single-table real-time synchronization example

View the complete example

SET 'execution.checkpointing.interval' = '10s';

CREATE TABLE cdc_mysql_source (
  id int
  ,name VARCHAR
  ,PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = '<mysql-host>',
'port' = '3306',
'username' = '<source-user>',
'password' = '<password>',
'database-name' = 'database',
'table-name' = 'table'
);

-- Supports synchronization of INSERT / UPDATE / DELETE events
CREATE TABLE doris_sink (
id INT,
name STRING
)
WITH (
  'connector' = 'doris',
  'fenodes' = '<warehouse-http-host>:<http-port>',
  'table.identifier' = 'database.table',
  'username' = '<warehouse-user>',
  'password' = '<password>',
  'sink.properties.format' = 'json',
  'sink.properties.read_json_by_line' = 'true',
  'sink.enable-delete' = 'true',  -- Synchronize delete events
  'sink.label-prefix' = 'doris_label'
);

insert into doris_sink select id,name from cdc_mysql_source;

Whole-database / multi-table synchronization

For whole-database or multi-table synchronization from a TP database, the Flink Doris Connector provides an out-of-the-box whole-database sync feature that can create target tables and copy data for supported sources:

"<flink-home>/bin/flink" run \
    -Dexecution.checkpointing.interval=10s \
    -Dparallelism.default=1 \
    -c org.apache.doris.flink.tools.cdc.CdcTools \
    lib/flink-doris-connector-1.16-24.0.1.jar \
    mysql-sync-database \
    --database test_db \
    --mysql-conf hostname="<mysql-host>" \
    --mysql-conf port=3306 \
    --mysql-conf username="<source-user>" \
    --mysql-conf password="<password>" \
    --mysql-conf database-name=mysql_db \
    --including-tables "tbl1|test.*" \
    --sink-conf fenodes="<warehouse-http-host>:<http-port>" \
    --sink-conf username="<warehouse-user>" \
    --sink-conf password="<password>" \
    --sink-conf jdbc-url="jdbc:mysql://<warehouse-host>:<sql-port>" \
    --sink-conf sink.label-prefix=label

For more details, see Flink Doris Connector.

---

Option 3: Spark connector

This option suits scenarios where you already use Spark for batch processing. You can use the Spark Connector's JDBC Source to read from the TP database and the Doris Sink to write into your warehouse.

val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:postgresql://<postgresql-host>:5432/<database>")
  .option("dbtable", "schema.tablename")
  .option("user", "username")
  .option("password", "password")
  .load()

 jdbcDF.write.format("doris")
  .option("doris.table.identifier", "db.table")
  .option("doris.fenodes", "<warehouse-http-host>:<http-port>")
  .option("user", "<warehouse-user>")
  .option("password", "<warehouse-password>")
  .save()

Related documents:

• JDBC To Other Databases
• Spark Doris Connector

---

Option 4: Streaming Job (built-in continuous synchronization in your warehouse)

VeloDB Cloud provides a built-in Streaming Job capability that continuously synchronizes data from MySQL or PostgreSQL without depending on an external compute engine.

Multi-table synchronization example

View the complete example

CREATE JOB multi_table_sync
ON STREAMING
FROM MYSQL (
        "jdbc_url" = "jdbc:mysql://<mysql-host>:3306",
        "driver_url" = "<jdbc-driver-url>",
        "driver_class" = "com.mysql.cj.jdbc.Driver",
        "user" = '<source-user>',
        "password" = '<password>',
        "database" = "test",
        "include_tables" = "user_info,order_info",
        "offset" = "initial"
)
TO DATABASE target_test_db;

For more details, see Postgres / MySQL Continuous Load.

---

Option 5: third-party synchronization tools

DataX is primarily a batch transfer tool. CDC and whole-database support depend on the tool, source, and connector version.

In addition to the native options above, you can use common third-party synchronization tools from the community. These suit scenarios where you already have a synchronization platform or want to manage tasks through a visual interface:

• NineData
• DataX
• SeaTunnel
• CloudCanal

---

FAQ

Q1: Which option should you choose for a one-time migration of a few MySQL tables to VeloDB Cloud?

Use Multi-Catalog. After creating a JDBC Catalog, run INSERT INTO or CTAS to complete the migration without any extra component.

Q2: How do you synchronize incremental data (including UPDATE / DELETE) from MySQL to VeloDB Cloud in real time?

Use Flink Doris Connector + Flink CDC and enable sink.enable-delete = true in the Doris Sink, or use the built-in Streaming Job in your warehouse for continuous synchronization.

Q3: How do you synchronize an entire MySQL database to VeloDB Cloud in one step?

Use the mysql-sync-database whole-database sync feature in the Flink Doris Connector, or the multi-table synchronization capability of Streaming Job. Both can create target tables and synchronize data for supported sources. Check the chosen mode's type mapping, key requirements, and supported schema changes before relying on automatic schema evolution.

Q4: Are Oracle, SQL Server, and PostgreSQL also supported as source databases?

Yes. Multi-Catalog and the Flink / Spark Connectors support TP databases such as Oracle, SQL Server, and PostgreSQL through their respective JDBC drivers. Streaming Job currently supports MySQL and PostgreSQL.

Q5: How do you specify the replication count for the VeloDB Cloud table during migration?

Replica allocation is managed by VeloDB Cloud. Omit self-managed replication_num, replication_allocation, and connector table-replication overrides in migration commands. These settings do not let you control the warehouse storage replica count.