通过Flink导入数据

功能简介

Flink Doris Connector是一个用于Apache Flink和Apache Doris之间的连接器，它允许用户在Flink中读取Doris数据和写入数据到Doris，从而为实时数据处理和分析提供了支持。由于SelectDB完全兼容Apache Doris，所以Flink Doris Connector也是SelectDB流式导入数据的常用方式。

Flink的流处理能力，基于Source、Transform和Sink三个组件。各个组件功能如下：

• Source（数据源）：

  • 作用：Source是Flink数据流的入口，用于从外部系统读取数据流。这些外部系统可能包括消息队列（如Apache Kafka）、数据库、文件系统等。

  • 示例：使用Kafka作为数据源读取实时消息，或者从文件中读取数据。

• Transform（数据转换）：

  • 作用：Transform阶段负责对输入的数据流进行处理和转换。这些转换操作可以是过滤、映射、聚合、窗口操作等。

  • 示例：对输入流进行映射操作，将输入的数据结构转换成另一种形式，或对数据进行聚合，计算每分钟的某个指标等。

• Sink（数据汇）：

  • 作用：Sink是Flink数据流的出口，用于将处理后的数据输出到外部系统。Sink可以将数据写入数据库、文件、消息队列等。

  • 示例：将处理后的结果写入MySQL数据库，或者将数据发送到另一个Kafka主题。

通过Flink Doris Connector将数据导入至SelectDB，各数据的流向如下图所示。

前提条件

• 数据源、Fink与SelectDB网络互通：

  1. 为云数据库 SelectDB 版实例申请公网地址。具体操作，请参见申请和释放公网地址。

     如果您的Flink与数据源均为阿里云产品，或者均部署在阿里云服务器上，且阿里云产品或阿里云服务器与云数据库 SelectDB 版实例位于同一VPC下，跳过此步骤。

  2. 将Flink以及数据源的相关IP添加至云数据库 SelectDB 版的白名单。具体操作，请参见设置白名单。

• Flink已引入与Flink Doris Connector。

  Flink与Flink Doris Connector版本要求如下：

  Flink版本	Flink Doris Connector版本	下载地址
  阿里云实时计算Flink版：大于或等于1.17 开源版：大于或等于1.15	1.5.2及以上版本	Flink Doris Connector

  Flink如何引入Flink Doris Connector，请参见如何引入Flink Doris Connector。

如何引入Flink Doris Connector

根据您的实际情况，引入Flink Doris Connector。

• 如果您的是通过阿里云实时计算Flink版导入数据至SelectDB，您可以通过自定义连接器来上传、使用和更新Flink Doris Connector。如何使用自定义连接器，请参见管理自定义连接器。

• 如果您的Flink集群是自建的开源版集群，您需要下载对应Flink Doris Connector版本的JAR包，并放置在Flink安装目录的lib目录下。JAR包下载地址，请参见JAR包。

• 如果您需要以Maven的方式引入Flink Doris Connector，需在项目的依赖配置文件中添加以下代码。更多版本，请参见Maven仓库。

  <!-- flink-doris-connector -->
  <dependency>
    <groupId>org.apache.doris</groupId>
    <artifactId>flink-doris-connector-1.16</artifactId>
    <version>1.5.2</version>
  </dependency>  

使用示例

环境准备

通过Flink SQL方式导入数据

1. 启动Flink SQL Client服务。

   在Flink安装目录的bin目录下，执行以下指令。

   ./sql-client.sh

2. 在Flink SQL Client上提交Flink任务，具体步骤如下。

   1. 创建MySQL源表。

      下述语句中，WITH后面的配置项为MySQL CDC Source的信息，配置项详情，请参见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. 创建SelectDB结果表。

      下述语句中，WITH后面的配置项为SelectDB的信息，配置项详情，请参见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. 同步MySql源表数据至SelectDB结果表。

      INSERT INTO employees_sink SELECT * FROM employees_source;

3. 查看数据导入结果。

   连接SelectDB，执行以下语句，查看数据导入结果。

   SELECT * FROM test.employees;

通过Flink CDC方式

以下介绍如何使用Flink CDC将数据库的数据导入至SelectDB。

在Flink安装目录下，通过flink程序执行Flink CDC，语法如下：

<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> ...]]

参数说明

Sink配置项

同步示例

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

通过DataStream方式

1. 在Maven项目中，引入相关依赖。

   相关Maven依赖

   <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. Java核心代码。

   下述代码中，MySql源表配置与SelectDB结果表配置的参数，与上述通过Flink SQL方式导入数据中的配置一一对应，详情请参见MySQL | Apache Flink CDC与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);
           
           // 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();
   
           // 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();
   
           // 配置Stream Load相关参数 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");
       }
   }

使用进阶

使用Flink SQL更新部分列数据

-- enable 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' -- 开启部分列更新
);


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

使用Flink SQL根据指定列删除数据

在数据源为CDC的场景中，Doris Sink会根据RowKind来区分事件的类型，对隐藏列__DORIS_DELETE_SIGN__进行赋值以达到删除的目的。在数据源为Kafka消息的场景中，Doris Sink无法直接使用RowKind来区分操作类型，需要依赖消息中的特定字段来标记操作类型，比如{"op_type":"delete",data:{...}}，针对这类数据，希望将op_type=delete的数据删除掉。此时需要根据业务逻辑判断，显式地传入隐藏列的值。下面以Flink SQL方式为例，介绍如何根据Kafka数据中的特定字段删除SelectDB中的数据。

-- 比如数据: {"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表示不从RowKind获取事件类型
  'sink.properties.columns' = 'id, name, __DORIS_DELETE_SIGN__'  -- 显示指定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;

常见问题

• Q：如何写入Bitmap类型？

  A：示例如下所示：

  CREATE TABLE bitmap_sink (
    dt INT,
    page STRING,
    user_id INT 
  )
  WITH ( 
    'connector' = 'doris', 
    'fenodes' = 'selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080',
    'table.identifier' = 'test.bitmap_test', 
    'username' = 'admin', 
    'password' = '****', 
    'sink.label-prefix' = 'selectdb_label', 
    'sink.properties.columns' = 'dt,page,user_id,user_id=to_bitmap(user_id)'
  );

• Q：如何解决报错：errCode = 2, detailMessage = Label[label_0_1]has already been used, relate to txn[19650]。

  A：Exactly-Once场景下，Flink Job重启时必须从最新的Checkpoint或Savepoint启动，否则会报如上错误。不要求Exactly-Once时，也可通过关闭两阶段提交（2PC）sink.enable-2pc=false 或更换不同的sink.label-prefix解决。

• Q：如何解决报错：errCode = 2, detailMessage = transaction[19650]not found。

  A：此报错发生在提交（Commit）阶段，Checkpoint里面记录的事务ID，在SelectDB侧已经过期，此时再次提交（Commit）就会出现上述错误。 此时无法从Checkpoint启动，后续可通过修改SelectDB的参数streaming_label_keep_max_second配置来延长过期时间，默认为12小时。

• Q：如何解决报错：errCode = 2, detailMessage = current running txns on db 10006 is 100, larger than limit 100。

  A：此报错是因为同一个库并发导入超过了100，可以通过调整SelectDB的参数max_running_txn_num_per_db来解决，详情请参见max_running_txn_num_per_db。

  同时，一个任务频繁修改label重启，也可能会导致这个错误。两阶段提交（2PC）场景下（Duplicate/Aggregate模型），每个任务的label需要唯一，并且从Checkpoint重启时，Flink任务才会主动中止（abort）之前启动的但未完成（即已precommit但未Commit的）事务（txn）。如果频繁修改label重启，会导致大量precommit成功的事务（txn）无法被中止（abort），占用事务。在Unique模型下也可关闭两阶段提交（2PC），通过设计Sink来实现幂等写入。

• Q：Flink写入Unique模型时，如何保证一批数据的有序性？

  A：可以添加sequence列配置来保证，更多详情，请参见sequence。

• Q：为什么Flink任务没报错，但是无法同步数据？

  A：Connector 1.1.0版本以前，是攒批写入的，写入均是由数据驱动，需要判断上游是否有数据写入。1.1.0之后，依赖Checkpoint，必须开启Checkpoint才能写入。

• Q：如何解决报错：tablet writer write failed, tablet_id=190958, txn_id=3505530, err=-235。

  A：此报错通常发生在Connector 1.1.0版本之前，是由于写入频率过快，导致版本过多。可以通过设置sink.buffer-flush.max-bytes和sink.buffer-flush.interval参数来降低Stream Load的频率。

• Q：Flink导入时有脏数据，如何跳过？

  A：Flink在数据导入时，如果有脏数据，比如字段格式、长度等问题，会导致Stream Load报错，此时Flink会不断地重试。如果需要跳过，可以通过禁用Stream Load的严格模式（strict_mode=false,max_filter_ratio=1）或者在Sink算子之前对数据做过滤。

• Q：源表和SelectDB表应如何对应？

  A：使用Flink Doris Connector导入数据时，要注意两个方面，一是源表的列和类型要跟Flink SQL中的列和类型对应；二是Flink SQL中的列和类型要跟SelectDB表的列和类型对应。

• Q：如何解决报错：TApplicationException: get_next failed: out of sequence response: expected 4 but got 3。

  A：此报错是由于Thrift框架存在并发bug导致的，建议您使用尽可能新的Connector以及与之兼容的Flink版本。

• Q：如何解决报错：DorisRuntimeException: Fail to abort transaction 26153 with urlhttp://192.168.XX.XX。

  A：你可以在TaskManager中搜索日志abort transaction response，根据HTTP返回码确定是客户端（Client）的问题还是服务器（Server）的问题。