Import data using Flink

Version requirements

Set up the environment

Import data using Flink SQL

1. Start the Flink SQL Client from the bin directory.

   ./sql-client.sh

2. Create the MySQL source table. The WITH clause configures the MySQL Change Data Capture (CDC) source. For all available options, see MySQL CDC connector options.

   CREATE TABLE employees_source (
       emp_no INT,
       birth_date DATE,
       first_name STRING,
       last_name STRING,
       gender STRING,
       hire_date DATE,
       PRIMARY KEY (`emp_no`) NOT ENFORCED
   ) WITH (
       'connector' = 'mysql-cdc',
       'hostname' = '127.0.0.1',
       'port' = '3306',
       'username' = 'root',
       'password' = '****',
       'database-name' = 'test',
       'table-name' = 'employees'
   );

3. Create the SelectDB sink table. The WITH clause configures the Doris sink. For all available parameters, see Sink configuration parameters.

   CREATE TABLE employees_sink (
       emp_no       INT,
       birth_date   DATE,
       first_name   STRING,
       last_name    STRING,
       gender       STRING,
       hire_date    DATE
   )
   WITH (
     'connector' = 'doris',
     'fenodes' = 'selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080',
     'table.identifier' = 'test.employees',
     'username' = 'admin',
     'password' = '****'
   );

4. Synchronize data from MySQL to SelectDB.

   INSERT INTO employees_sink SELECT * FROM employees_source;

5. Verify the import. Connect to SelectDB and run:

   SELECT * FROM test.employees;

Import data using Flink CDC

Flink CDC synchronizes an entire database from a source to SelectDB in one command. Run the following from your Flink installation directory:

<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-1.5.2.jar \
    <mysql-sync-database|oracle-sync-database|postgres-sync-database|sqlserver-sync-database> \
    --database <selectdb-database-name> \
    [--job-name <flink-job-name>] \
    [--table-prefix <selectdb-table-prefix>] \
    [--table-suffix <selectdb-table-suffix>] \
    [--including-tables <mysql-table-name|name-regular-expr>] \
    [--excluding-tables <mysql-table-name|name-regular-expr>] \
    --mysql-conf <mysql-cdc-source-conf> [--mysql-conf <mysql-cdc-source-conf> ...] \
    --oracle-conf <oracle-cdc-source-conf> [--oracle-conf <oracle-cdc-source-conf> ...] \
    --sink-conf <doris-sink-conf> [--sink-conf <doris-sink-conf> ...] \
    [--table-conf <selectdb-table-conf> [--table-conf <selectdb-table-conf> ...]]

Add the required CDC JAR (e.g., flink-sql-connector-mysql-cdc-${version}.jar) to $FLINK_HOME/lib before running.

Full database synchronization requires Flink 1.15 or later.

<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-1.5.2.jar \
    mysql-sync-database \
    --database test_db \
    --mysql-conf hostname=127.0.0.1 \
    --mysql-conf port=3306 \
    --mysql-conf username=root \
    --mysql-conf password="password" \
    --mysql-conf database-name=test \
    --including-tables "tbl1|test.*" \
    --sink-conf fenodes=selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080 \
    --sink-conf username=admin \
    --sink-conf password=****

<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-1.5.2.jar \
    oracle-sync-database \
    --database test_db \
    --oracle-conf hostname=127.0.0.1 \
    --oracle-conf port=1521 \
    --oracle-conf username=admin \
    --oracle-conf password="password" \
    --oracle-conf database-name=XE \
    --oracle-conf schema-name=ADMIN \
    --including-tables "tbl1|test.*" \
    --sink-conf fenodes=selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080 \
    --sink-conf username=admin \
    --sink-conf password=****

<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-1.5.2.jar \
    postgres-sync-database \
    --database db1 \
    --postgres-conf hostname=127.0.0.1 \
    --postgres-conf port=5432 \
    --postgres-conf username=postgres \
    --postgres-conf password="123456" \
    --postgres-conf database-name=postgres \
    --postgres-conf schema-name=public \
    --postgres-conf slot.name=test \
    --postgres-conf decoding.plugin.name=pgoutput \
    --including-tables "tbl1|test.*" \
    --sink-conf fenodes=selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080 \
    --sink-conf username=admin \
    --sink-conf password=****

<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-1.5.2.jar \
    sqlserver-sync-database \
    --database db1 \
    --sqlserver-conf hostname=127.0.0.1 \
    --sqlserver-conf port=1433 \
    --sqlserver-conf username=sa \
    --sqlserver-conf password="123456" \
    --sqlserver-conf database-name=CDC_DB \
    --sqlserver-conf schema-name=dbo \
    --including-tables "tbl1|test.*" \
    --sink-conf fenodes=selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080 \
    --sink-conf username=admin \
    --sink-conf password=****

Import data using the DataStream API

1. Add the required Maven dependencies to your pom.xml.

   <properties>
       <maven.compiler.source>8</maven.compiler.source>
       <maven.compiler.target>8</maven.compiler.target>
       <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
       <scala.version>2.12</scala.version>
       <java.version>1.8</java.version>
       <flink.version>1.16.3</flink.version>
       <fastjson.version>1.2.62</fastjson.version>
       <scope.mode>compile</scope.mode>
   </properties>
   <dependencies>
       <dependency>
           <groupId>com.google.guava</groupId>
           <artifactId>guava</artifactId>
           <version>28.1-jre</version>
       </dependency>
       <dependency>
           <groupId>org.apache.commons</groupId>
           <artifactId>commons-lang3</artifactId>
           <version>3.14.0</version>
       </dependency>
       <!-- Flink Doris Connector -->
       <dependency>
           <groupId>org.apache.doris</groupId>
           <artifactId>flink-doris-connector-1.16</artifactId>
           <version>1.5.2</version>
       </dependency>
       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-table-api-scala-bridge_${scala.version}</artifactId>
           <version>${flink.version}</version>
       </dependency>
       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-table-planner_${scala.version}</artifactId>
           <version>${flink.version}</version>
       </dependency>
       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-streaming-scala_${scala.version}</artifactId>
           <version>${flink.version}</version>
       </dependency>
       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-clients</artifactId>
           <version>${flink.version}</version>
       </dependency>
       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-connector-jdbc</artifactId>
           <version>${flink.version}</version>
       </dependency>
       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-connector-kafka</artifactId>
           <version>${flink.version}</version>
       </dependency>
       <dependency>
           <groupId>com.ververica</groupId>
           <artifactId>flink-sql-connector-mysql-cdc</artifactId>
           <version>2.4.2</version>
           <exclusions>
               <exclusion>
                   <artifactId>flink-shaded-guava</artifactId>
                   <groupId>org.apache.flink</groupId>
               </exclusion>
           </exclusions>
       </dependency>
       <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-runtime-web</artifactId>
           <version>${flink.version}</version>
       </dependency>
   </dependencies>

2. Write the job code. The MySQL source parameters and SelectDB sink parameters correspond to the configurations used in the Flink SQL example. For MySQL source options, see MySQL CDC connector options. For sink options, see Sink configuration parameters.

   package org.example;
   
   import com.ververica.cdc.connectors.mysql.source.MySqlSource;
   import com.ververica.cdc.connectors.mysql.table.StartupOptions;
   import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.json.JsonConverterConfig;
   import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
   
   import org.apache.doris.flink.cfg.DorisExecutionOptions;
   import org.apache.doris.flink.cfg.DorisOptions;
   import org.apache.doris.flink.sink.DorisSink;
   import org.apache.doris.flink.sink.writer.serializer.JsonDebeziumSchemaSerializer;
   import org.apache.doris.flink.tools.cdc.mysql.DateToStringConverter;
   import org.apache.flink.api.common.eventtime.WatermarkStrategy;
   import org.apache.flink.streaming.api.datastream.DataStreamSource;
   import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
   
   import java.util.HashMap;
   import java.util.Map;
   import java.util.Properties;
   
   public class Main {
       public static void main(String[] args) throws Exception {
           StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
           env.setParallelism(1);
           env.enableCheckpointing(10000);
   
           Map<String, Object> customConverterConfigs = new HashMap<>();
           customConverterConfigs.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, "numeric");
           JsonDebeziumDeserializationSchema schema =
                   new JsonDebeziumDeserializationSchema(false, customConverterConfigs);
   
           // Configure the MySQL source
           MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                   .hostname("rm-xxx.mysql.rds.aliyuncs***")
                   .port(3306)
                   .startupOptions(StartupOptions.initial())
                   .databaseList("db_test")
                   .tableList("db_test.employees")
                   .username("root")
                   .password("test_123")
                   .debeziumProperties(DateToStringConverter.DEFAULT_PROPS)
                   .deserializer(schema)
                   .serverTimeZone("Asia/Shanghai")
                   .build();
   
           // Configure the SelectDB sink
           DorisSink.Builder<String> sinkBuilder = DorisSink.builder();
           DorisOptions.Builder dorisBuilder = DorisOptions.builder();
           dorisBuilder.setFenodes("selectdb-cn-xxx-public.selectdbfe.rds.aliyunc****:8080")
                   .setTableIdentifier("db_test.employees")
                   .setUsername("admin")
                   .setPassword("test_123");
           DorisOptions dorisOptions = dorisBuilder.build();
   
           // Configure Stream Load parameters
           Properties properties = new Properties();
           properties.setProperty("format", "json");
           properties.setProperty("read_json_by_line", "true");
           DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
           executionBuilder.setStreamLoadProp(properties);
   
           sinkBuilder.setDorisExecutionOptions(executionBuilder.build())
                   .setSerializer(JsonDebeziumSchemaSerializer.builder().setDorisOptions(dorisOptions).build())
                   .setDorisOptions(dorisOptions);
   
           DataStreamSource<String> dataStreamSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");
           dataStreamSource.sinkTo(sinkBuilder.build());
           env.execute("MySQL to SelectDB");
       }
   }

Update partial columns

Use sink.properties.partial_columns = 'true' to update only a subset of columns in an existing row.

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

CREATE TABLE cdc_mysql_source (
   id INT,
   name STRING,
   bank STRING,
   age INT,
   PRIMARY KEY (id) NOT ENFORCED
) WITH (
 'connector' = 'mysql-cdc',
 'hostname' = '127.0.0.1',
 'port' = '3306',
 'username' = 'root',
 'password' = 'password',
 'database-name' = 'database',
 'table-name' = 'table'
);

CREATE TABLE selectdb_sink (
    id INT,
    name STRING,
    bank STRING,
    age INT
)
WITH (
  'connector' = 'doris',
  'fenodes' = 'selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080',
  'table.identifier' = 'database.table',
  'username' = 'admin',
  'password' = '****',
  'sink.properties.format' = 'json',
  'sink.properties.read_json_by_line' = 'true',
  'sink.properties.columns' = 'id,name,bank,age',
  'sink.properties.partial_columns' = 'true'  -- Enable partial column updates
);

INSERT INTO selectdb_sink SELECT id, name, bank, age FROM cdc_mysql_source;

Delete rows based on a Kafka field

In a CDC source, the Doris sink uses RowKind to identify delete events automatically. In a Kafka source, there is no RowKind, so you must map a message field to the hidden column __DORIS_DELETE_SIGN__ to trigger deletions.

-- Example Kafka message: {"op_type":"delete","data":{"id":1,"name":"zhangsan"}}
CREATE TABLE KAFKA_SOURCE(
  data STRING,
  op_type STRING
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE SELECTDB_SINK(
  id INT,
  name STRING,
  __DORIS_DELETE_SIGN__ INT
) WITH (
  'connector' = 'doris',
  'fenodes' = 'selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080',
  'table.identifier' = 'db.table',
  'username' = 'admin',
  'password' = '****',
  'sink.enable-delete' = 'false',                                    -- Do not use RowKind for delete detection
  'sink.properties.columns' = 'id, name, __DORIS_DELETE_SIGN__'     -- Specify columns explicitly
);

INSERT INTO SELECTDB_SINK
SELECT json_value(data,'$.id') as id,
       json_value(data,'$.name') as name,
       if(op_type='delete', 1, 0) as __DORIS_DELETE_SIGN__
FROM KAFKA_SOURCE;