ApsaraDB for SelectDB：Impor data menggunakan Flink

Lingkungan contoh

Bagian ini memberikan contoh cara menggunakan Flink SQL, Flink CDC, dan DataStream untuk memigrasikan data dari tabel `employees` di database `test` instans MySQL ke tabel `employees` di database `test` instans SelectDB. Anda dapat memodifikasi parameter sesuai kebutuhan. Lingkungan contohnya sebagai berikut:

• Lingkungan standalone Flink 1.16

• Java

• Database tujuan: test

• Tabel tujuan: employees

• Database sumber: test

• Tabel sumber: employees

Persiapkan lingkungan

Impor data menggunakan Flink SQL

1. Jalankan layanan Flink SQL Client.

   Jalankan perintah berikut di direktori `bin` instalasi Flink.

   ./sql-client.sh

2. Kirim pekerjaan Flink di Flink SQL Client.

   1. Buat tabel sumber MySQL.

      Dalam pernyataan berikut, item dalam klausa `WITH` adalah konfigurasi untuk MySQL CDC Source. Untuk informasi selengkapnya tentang item konfigurasi, lihat MySQL | Apache Flink CDC.

      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'
      );

   2. Buat tabel sink SelectDB.

      Dalam pernyataan berikut, item dalam klausa `WITH` adalah konfigurasi untuk SelectDB. Untuk informasi selengkapnya tentang item konfigurasi, lihat Item konfigurasi sink.

      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' = '****'
      );

   3. Sinkronkan data dari tabel sumber MySQL ke tabel sink SelectDB.

      INSERT INTO employees_sink SELECT * FROM employees_source;

3. Lihat hasil impor data.

   Hubungkan ke SelectDB dan jalankan pernyataan berikut untuk melihat hasilnya.

   SELECT * FROM test.employees;

Impor data menggunakan Flink CDC

Bagian ini menjelaskan cara menggunakan Flink CDC untuk mengimpor data dari database ke SelectDB.

Di direktori instalasi Flink, Anda dapat menggunakan program flink untuk menjalankan Flink CDC. Sintaksnya sebagai berikut:

<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> [--table-conf <doris-sink-conf> ...] \
    [--table-conf <selectdb-table-conf> [--table-conf <selectdb-table-conf> ...]]

<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 \
    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=****

Impor data menggunakan API DataStream

1. Dalam proyek Maven Anda, tambahkan dependensi yang diperlukan.

   Dependensi Maven yang diperlukan

   <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>
           <!-- https://mvnrepository.com/artifact/com.google.guava/guava -->
           <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>
   
           <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>org.apache.doris</groupId>
               <artifactId>flink-doris-connector-1.16</artifactId>
               <version>1.5.2</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. Kode inti Java.

   Dalam kode berikut, parameter untuk tabel sumber MySQL dan tabel sink SelectDB sesuai dengan konfigurasi di bagian Impor data menggunakan Flink SQL. Untuk informasi selengkapnya, lihat MySQL | Apache Flink CDC dan Item konfigurasi sink.

   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);
           
           // Konfigurasikan tabel sumber MySQL.
           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();
   
           // Konfigurasikan tabel sink SelectDB.
           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();
   
           // Konfigurasikan parameter Stream Load di sink.properties.
           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()) //serialize according to string
                   .setDorisOptions(dorisOptions);
   
           DataStreamSource<String> dataStreamSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");
           dataStreamSource.sinkTo(sinkBuilder.build());
           env.execute("MySQL to SelectDB");
       }
   }

Perbarui kolom parsial menggunakan Flink SQL

-- aktifkan checkpoint
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' -- Aktifkan pembaruan kolom parsial.
);


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

Hapus data berdasarkan kolom tertentu menggunakan Flink SQL

Dalam skenario sumber CDC, Doris Sink membedakan jenis event berdasarkan `RowKind` dan memberikan nilai ke kolom tersembunyi __DORIS_DELETE_SIGN__ untuk melakukan penghapusan. Dalam skenario sumber pesan Kafka, Doris Sink tidak dapat langsung menggunakan `RowKind` untuk membedakan jenis operasi. Komponen ini harus mengandalkan bidang tertentu dalam pesan untuk menandai jenis operasi, seperti {"op_type":"delete",data:{...}}. Untuk data jenis ini, jika Anda ingin menghapus catatan di mana `op_type` bernilai 'delete', Anda harus secara eksplisit meneruskan nilai untuk kolom tersembunyi berdasarkan logika bisnis Anda. Contoh Flink SQL berikut menunjukkan cara menghapus data di SelectDB berdasarkan bidang tertentu dalam data Kafka.

-- Data contoh: {"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',        -- false menunjukkan bahwa jenis event tidak diperoleh dari RowKind.
  'sink.properties.columns' = 'id, name, __DORIS_DELETE_SIGN__'  -- Secara eksplisit tentukan kolom yang akan diimpor untuk Stream Load.
);

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;