大数据系列之：Flink Doris Connector，实时同步数据到Doris数据库_两个doris 库 同步

可以通过Flink操作（读取、插入、修改、删除）支持存储在Doris中的数据。本文介绍了如何通过Datastream和Flink操作Doris。

注意：

• 修改和删除仅支持唯一键模型。
• 当前的删除是为了支持Flink CDC访问数据以实现自动删除。如果要删除其他数据访问方法，您需要自行实现。

一、版本兼容性

二、使用

Maven

添加 flink-doris-connector

<!-- flink-doris-connector -->
<dependency>
   <groupId>org.apache.doris</groupId>
   <artifactId>flink-doris-connector-1.16</artifactId>
   <version>1.6.0</version>
</dependency>
1
2
3
4
5
6

• 请根据不同的Flink版本替换相应的Connector和Flink依赖版本。
• 也可以从这里下载相关版本的jar包。

flink-doris-connector下载地址：

• https://repo.maven.apache.org/maven2/org/apache/doris/

编译

• 编译时直接运行sh build.sh即可。
• 编译成功后，会在dist目录下生成目标jar包，如：flink-doris-connector-1.5.0-SNAPSHOT.jar。将此文件复制到 Flink 的类路径中以使用 Flink-Doris-Connector。例如，Flink 运行在 Local 模式，则将此文件放在 lib/ 文件夹中。 Flink运行在Yarn集群模式下，将此文件放入预部署包中。

三、Flink SQL

read

-- doris source
CREATE TABLE flink_doris_source (
     name STRING,
     age INT,
     price DECIMAL(5,2),
     sale DOUBLE
     )
     WITH (
       'connector' = 'doris',
       'fenodes' = 'FE_IP:HTTP_PORT',
       'table.identifier' = 'database.table',
       'username' = 'root',
       'password' = 'password'
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14

write

--enable checkpoint
SET 'execution.checkpointing.interval' = '10s';

-- doris sink
CREATE TABLE flink_doris_sink (
     name STRING,
     age INT,
     price DECIMAL(5,2),
     sale DOUBLE
     )
     WITH (
       'connector' = 'doris',
       'fenodes' = 'FE_IP:HTTP_PORT',
       'table.identifier' = 'db.table',
       'username' = 'root',
       'password' = 'password',
       'sink.label-prefix' = 'doris_label'
);

-- submit insert job
INSERT INTO flink_doris_sink select name,age,price,sale from flink_doris_source
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

四、DataStream

read

DorisOptions.Builder builder = DorisOptions.builder()
         .setFenodes("FE_IP:HTTP_PORT")
         .setTableIdentifier("db.table")
         .setUsername("root")
         .setPassword("password");

DorisSource<List<?>> dorisSource = DorisSource.<List<?>>builder()
         .setDorisOptions(builder.build())
         .setDorisReadOptions(DorisReadOptions.builder().build())
         .setDeserializer(new SimpleListDeserializationSchema())
         .build();

env.fromSource(dorisSource, WatermarkStrategy.noWatermarks(), "doris source").print();
1
2
3
4
5
6
7
8
9
10
11
12
13

write

DorisSink通过StreamLoad向Doris写入数据，DataStream写入时支持不同的序列化方式

字符串数据流（SimpleStringSerializer）

// enable checkpoint
env.enableCheckpointing(10000);
// using batch mode for bounded data
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

DorisSink.Builder<String> builder = DorisSink.builder();
DorisOptions.Builder dorisBuilder = DorisOptions.builder();
dorisBuilder.setFenodes("FE_IP:HTTP_PORT")
         .setTableIdentifier("db.table")
         .setUsername("root")
         .setPassword("password");

Properties properties = new Properties();
// When the upstream is writing json, the configuration needs to be enabled.
//properties.setProperty("format", "json");
//properties.setProperty("read_json_by_line", "true");
DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
executionBuilder.setLabelPrefix("label-doris") //streamload label prefix
                 .setDeletable(false)
                 .setStreamLoadProp(properties); ;

builder.setDorisReadOptions(DorisReadOptions.builder().build())
         .setDorisExecutionOptions(executionBuilder.build())
         .setSerializer(new SimpleStringSerializer()) //serialize according to string
         .setDorisOptions(dorisBuilder.build());

//mock string source
List<Tuple2<String, Integer>> data = new ArrayList<>();
data.add(new Tuple2<>("doris",1));
DataStreamSource<Tuple2<String, Integer>> source = env. fromCollection(data);

source.map((MapFunction<Tuple2<String, Integer>, String>) t -> t.f0 + "\t" + t.f1)
       .sinkTo(builder.build());

//mock json string source
//env.fromElements("{\"name\":\"zhangsan\",\"age\":1}").sinkTo(builder.build());
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

RowData数据流（RowDataSerializer）

// enable checkpoint
env.enableCheckpointing(10000);
// using batch mode for bounded data
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

//doris sink option
DorisSink.Builder<RowData> builder = DorisSink.builder();
DorisOptions.Builder dorisBuilder = DorisOptions.builder();
dorisBuilder.setFenodes("FE_IP:HTTP_PORT")
         .setTableIdentifier("db.table")
         .setUsername("root")
         .setPassword("password");

// json format to streamload
Properties properties = new Properties();
properties.setProperty("format", "json");
properties.setProperty("read_json_by_line", "true");
DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
executionBuilder.setLabelPrefix("label-doris") //streamload label prefix
                 .setDeletable(false)
                 .setStreamLoadProp(properties); //streamload params

//flink rowdata's schema
String[] fields = {"city", "longitude", "latitude", "destroy_date"};
DataType[] types = {DataTypes.VARCHAR(256), DataTypes.DOUBLE(), DataTypes.DOUBLE(), DataTypes.DATE()};

builder.setDorisReadOptions(DorisReadOptions.builder().build())
         .setDorisExecutionOptions(executionBuilder.build())
         .setSerializer(RowDataSerializer.builder() //serialize according to rowdata
                            .setFieldNames(fields)
                            .setType("json") //json format
                            .setFieldType(types).build())
         .setDorisOptions(dorisBuilder.build());

//mock rowdata source
DataStream<RowData> source = env. fromElements("")
     .map(new MapFunction<String, RowData>() {
         @Override
         public RowData map(String value) throws Exception {
             GenericRowData genericRowData = new GenericRowData(4);
             genericRowData.setField(0, StringData.fromString("beijing"));
             genericRowData.setField(1, 116.405419);
             genericRowData.setField(2, 39.916927);
             genericRowData.setField(3, LocalDate.now().toEpochDay());
             return genericRowData;
         }
     });

source. sinkTo(builder. build());
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

SchemaChange数据流（JsonDebeziumSchemaSerializer）

// enable checkpoint
env.enableCheckpointing(10000);

Properties props = new Properties();
props. setProperty("format", "json");
props.setProperty("read_json_by_line", "true");
DorisOptions dorisOptions = DorisOptions. builder()
         .setFenodes("127.0.0.1:8030")
         .setTableIdentifier("test.t1")
         .setUsername("root")
         .setPassword("").build();

DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
executionBuilder.setLabelPrefix("label-prefix")
         .setStreamLoadProp(props).setDeletable(true);

DorisSink.Builder<String> builder = DorisSink.builder();
builder.setDorisReadOptions(DorisReadOptions.builder().build())
         .setDorisExecutionOptions(executionBuilder.build())
         .setDorisOptions(dorisOptions)
         .setSerializer(JsonDebeziumSchemaSerializer.builder().setDorisOptions(dorisOptions).build());

env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
         .sinkTo(builder.build());
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

五、Lookup Join

CREATE TABLE fact_table (
  `id` BIGINT,
  `name` STRING,
  `city` STRING,
  `process_time` as proctime()
) WITH (
  'connector' = 'kafka',
  ...
);

create table dim_city(
  `city` STRING,
  `level` INT ,
  `province` STRING,
  `country` STRING
) WITH (
  'connector' = 'doris',
  'fenodes' = '127.0.0.1:8030',
  'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
  'table.identifier' = 'dim.dim_city',
  'username' = 'root',
  'password' = ''
);

SELECT a.id, a.name, a.city, c.province, c.country,c.level 
FROM fact_table a
LEFT JOIN dim_city FOR SYSTEM_TIME AS OF a.process_time AS c
ON a.city = c.city
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

• 这个命令是用于创建两个表和一个查询语句。第一个表是名为"fact_table"的表，它有四个列，分别为"id"、“name”、“city"和"process_time”。其中，"id"列是BIGINT类型，"name"和"city"列是STRING类型，"process_time"列是一个基于当前系统时间的计算列，它使用"proctime()"函数实现。
• 第二个表是名为"dim_city"的表，它有四个列，分别为"city"、“level”、“province"和"country”。其中，“city”、“province"和"country"列是STRING类型，“level"列是INT类型。该表使用"Doris"作为存储引擎，连接器为"connector”，并且需要指定连接器的其他参数，如"fenodes”、“jdbc-url”、“table.identifier”、"username"和"password"等。
• 最后一个命令是一个查询语句，它使用"LEFT JOIN"将"fact_table"和"dim_city"两个表进行连接，并使用"FOR SYSTEM_TIME AS OF"来指定连接时的时间戳，这里使用了"process_time"列的值。查询结果包括"id"、“name”、“city”、“province”、"country"和"level"这些列。

六、配置

通用配置项

fenodes：

• Doris FE http地址，支持多个地址，用逗号分隔

benodes：

• Doris BE http地址，支持多个地址，以逗号分隔。

jdbc-url：

• jdbc连接信息，如：jdbc:mysql://127.0.0.1:9030

table.identifier：

• Doris表名，如：db.tbl

auto-redirect：

• 默认值：true
• 是否重定向 StreamLoad 请求。开启后StreamLoad会通过FE写入，不再显示BE信息。

doris.request.retries：

• 默认值：3
• 向 Doris 发送请求的重试次数

doris.request.connect.timeout.ms：

• 默认值：30000
• 向 Doris 发送请求的连接超时

doris.request.read.timeout.ms：

• 默认值：30000
• 读取向 Doris 发送请求的超时

源配置项

doris.request.query.timeout.s：

• 默认值：3600
• 查询Doris的超时时间，默认1小时，-1表示无超时限制

doris.request.tablet.size：

• 默认值：Integer. MAX_VALUE
• 一个Partition对应的Doris Tablet数量。该值设置得越小，生成的Partition就越多。这提高了 Flink 端的并行度，但同时也给 Doris 带来了更大的压力。

doris.batch.size：

• 默认值：1024
• 一次从BE读取数据的最大行数。增加该值会减少 Flink 和 Doris 之间建立的连接数量。从而减少网络延迟带来的额外时间开销。

doris.exec.mem.limit：

• 默认值：2147483648
• 单个查询的内存限制。默认为2GB，以字节为单位

doris.deserialize.arrow.async：

• 默认值：FALSE
• 是否支持flink-doris-connector迭代所需的Arrow格式异步转换为RowBatch

doris.deserialize.queue.size：

• 默认值：64
• Arrow格式的内部处理队列的异步转换，当doris.deserialize.arrow.async为true时有效

doris.read.field：

• 读取Doris表的列名列表，以逗号分隔

doris.filter.query：

• 过滤读取数据的表达式，这个表达式透明传递给Doris。 Doris使用这个表达式来完成源端的数据过滤。例如年龄=18。

接收器配置项

sink.label-prefix：

• Stream加载导入使用的标签前缀。在2pc场景下，需要全局唯一性来保证Flink的EOS语义。

sink.properties.*：

• 导入流负载参数。
  例如： ‘sink.properties.column_separator’ = ', ’ 定义列分隔符， ‘sink.properties.escape_delimiters’ = ‘true’ 特殊字符作为分隔符， ‘\x01’ 将转换为二进制 0x01
• JSON格式导入
  ‘sink.properties.format’ = ‘json’ ‘sink.properties.按行读取 json’ = ‘true’
  详细参数请参考这里。

sink.enable-delete：

• 默认值：TRUE
• 是否启用删除。该选项需要Doris表开启批量删除功能（Doris 0.15+版本默认开启），且仅支持Unique模型。

sink.enable-2pc：

• 默认值：TRUE
• 是否启用两阶段提交（2pc），默认为true，以保证Exactly-Once语义。

sink.buffer-size：

• 默认值：1MB
• 写入数据缓存缓冲区的大小，以字节为单位。不建议修改，默认配置即可

sink.buffer-count：

• 默认值：3
• 写入数据缓冲区的数量。不建议修改，默认配置即可

sink.max-retries：

• 默认值：3
• Commit失败后最大重试次数，默认3

sink.use-cache：

• 默认值：false
• 发生异常时，是否使用内存缓存进行恢复。启用后，Checkpoint 期间的数据将保留在缓存中。

sink.enable.batch-mode：

• 默认值：false
• 是否使用批处理模式写入Doris。使能后，写入时序不依赖于Checkpoint。写入是通过sink.buffer-flush.max-rows/sink.buffer-flush.max-bytes/sink.buffer-flush.interval参数控制的。输入机会。
  同时开启后，Exactly-once语义将无法保证。 Uniq模型可以用来实现幂等性。

sink.flush.queue-size：

• 默认值：2
• 在批处理模式下，缓存的列大小。

sink.buffer-flush.max-rows：

• 默认值：50000
• 批处理模式下，单批写入的最大数据行数。

sink.buffer-flush.max-bytes：

• 默认值：10MB
• 在批处理模式下，单批写入的最大字节数。

sink.buffer-flush.interval：

• 默认值：10s
• 批处理模式下，异步刷新缓存的时间间隔

sink.ignore.update-before：

• 默认值：true
• 是否忽略update-before事件，默认忽略。

查找Join配置项

lookup.cache.max-rows

• 默认值：-1
• 查找缓存的最大行数，默认值为-1，不启用缓存

lookup.cache.ttl：

• 默认值：10s
• 查找缓存的最大时间，默认10s

lookup.max-retries：

• 默认值：1
• 查找查询失败后重试的次数

lookup.jdbc.async：

• 默认值：false
• 是否启用异步查找，默认为false

lookup.jdbc.read.batch.size：

• 默认值：128
• 异步查找下，每个查询的最大批量大小

lookup.jdbc.read.batch.queue-size：

• 默认值：256
• 异步查找时中间缓冲队列的大小

lookup.jdbc.read.thread-size：

• 默认值：3
• 每个任务中用于查找的jdbc线程数

七、Doris 和 Flink 列类型映射

从connector-1.6.1版本开始，增加了对Variant、IPV6、IPV4三种数据类型读取的支持。读取 IPV6 和 Variant 需要 Doris 2.1.1 或更高版本。

八、使用Flink CDC访问Doris的示例

SET 'execution.checkpointing.interval' = '10s';
CREATE TABLE cdc_mysql_source (
  id int
  ,name VARCHAR
  ,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'
);

-- Support synchronous insert/update/delete events
CREATE TABLE doris_sink (
id INT,
name STRING
) 
WITH (
  'connector' = 'doris',
  'fenodes' = '127.0.0.1:8030',
  'table.identifier' = 'database.table',
  'username' = 'root',
  '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;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

九、使用FlinkSQL通过CDC访问并实现部分列更新的示例

-- 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 doris_sink (
    id INT,
    name STRING,
    bank STRING,
    age int
) 
WITH (
  'connector' = 'doris',
  'fenodes' = '127.0.0.1:8030',
  'table.identifier' = 'database.table',
  'username' = 'root',
  '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 doris_sink select id,name,bank,age from cdc_mysql_source;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

十、使用FlinkCDC访问多个表或整个数据库（支持MySQL、Oracle、PostgreSQL、SQLServer）

MySQL同步示例

<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.0-SNAPSHOT.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=123456 \
     --mysql-conf database-name=mysql_db \
     --including-tables "tbl1|test.*" \
     --sink-conf fenodes=127.0.0.1:8030 \
     --sink-conf username=root \
     --sink-conf password=123456 \
     --sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \
     --sink-conf sink.label-prefix=label \
     --table-conf replication_num=1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

Oracle同步示例

<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.0-SNAPSHOT.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|tbl2" \
      --sink-conf fenodes=127.0.0.1:8030 \
      --sink-conf username=root \
      --sink-conf password=\
      --sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \
      --sink-conf sink.label-prefix=label \
      --table-conf replication_num=1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

PostgreSQL 同步示例

<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.0-SNAPSHOT.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|tbl2" \
     --sink-conf fenodes=127.0.0.1:8030 \
     --sink-conf username=root \
     --sink-conf password=\
     --sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \
     --sink-conf sink.label-prefix=label \
     --table-conf replication_num=1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

SQLServer同步示例

<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.0-SNAPSHOT.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|tbl2" \
     --sink-conf fenodes=127.0.0.1:8030 \
     --sink-conf username=root \
     --sink-conf password=\
     --sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \
     --sink-conf sink.label-prefix=label \
     --table-conf replication_num=1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

十一、使用FlinkCDC更新Key列

一般来说，在业务数据库中，数字被用作表的主键，例如学生表中使用数字（id）作为主键，但随着业务的发展，与数据对应的数字可能会发生变化。在这种情况下，使用FlinkCDC + Doris Connector进行数据同步可以自动更新Doris主键列中的数据。

原理

Flink CDC的底层采集工具是Debezium。Debezium内部使用op字段来识别相应的操作：op字段的值为c、u、d和r，分别对应创建、更新、删除和读取。对于主键列的更新，FlinkCDC将向下游发送DELETE和INSERT事件，在数据同步到Doris后，会自动更新主键列的数据。

示例

Flink程序可以参考上述CDC同步示例。任务成功提交后，在MySQL端执行更新主键列语句（update student set id = ‘1002’ where id = ‘1001’），以修改Doris中的数据。

十二、使用Flink根据指定的列删除数据

通常，Kafka中的消息使用特定的字段来标记操作类型，例如{“op_type”:“delete”,data:{…}}。对于这种类型的数据，希望删除op_type=delete的数据。

默认情况下，DorisSink将根据RowKind来区分事件类型。通常，在cdc的情况下，可以直接获取事件类型，并将隐藏列__DORIS_DELETE_SIGN__赋值以实现删除的目的，而Kafka需要基于业务逻辑进行判断，显示传递给隐藏列的值。

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

CREATE TABLE DORIS_SINK(
  id INT,
  name STRING,
  __DORIS_DELETE_SIGN__ INT
) WITH (
  'connector' = 'doris',
  'fenodes' = '127.0.0.1:8030',
  'table.identifier' = 'db.table',
  'username' = 'root',
  'password' = '',
  'sink.enable-delete' = 'false',        -- false means not to get the event type from RowKind
  'sink.properties.columns' = 'id, name, __DORIS_DELETE_SIGN__'  -- Display the import column of the specified streamload
);

INSERT INTO DORIS_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;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

• 这段代码是一个示例，演示了如何使用Flink从Kafka源表读取数据，并将其写入Doris目标表。具体来说，如果源表中的数据op_type字段的值为"delete"，则希望在Doris目标表中删除相应的数据。
• 首先，在Kafka源表的定义中，我们有一个data字段用于存储源数据的JSON字符串，以及一个op_type字段用于标识操作类型。
• 然后，在Doris目标表的定义中，我们有一个id字段和一个name字段来存储数据的具体内容，还有一个名为__DORIS_DELETE_SIGN__的隐藏列，用于标识是否要进行删除操作。
• 在INSERT INTO语句中，我们将从Kafka源表中选择data字段的id和name，并使用json_value函数提取相应的值。同时，我们使用if函数将op_type字段的值与"delete"进行比较，如果相等则将__DORIS_DELETE_SIGN__赋值为1，否则赋值为0。
• 最后，将处理后的数据插入到Doris目标表中。
• 总之，这段代码的作用是根据源表中的op_type字段值，将对应的数据删除或写入到Doris目标表中。

十三、最佳实践应用场景

• 使用Flink Doris Connector最适合的场景是实时/批量将源数据同步到Doris（Mysql、Oracle、PostgreSQL）中，然后使用Flink对Doris和其他数据源中的数据进行联合分析。您还可以使用Flink Doris Connector。

其他注意事项：

• Flink Doris Connector主要依赖于Checkpoint进行流式写入，因此Checkpoint之间的时间间隔就是数据的可见延迟时间。
• 为了确保Flink的Exactly Once语义，Flink Doris Connector默认启用两阶段提交，Doris在1.1版本之后默认启用两阶段提交。1.0可以通过修改BE参数来启用。

十四、常见问题解答

Doris Source读取数据后，为什么流会结束？

• 目前Doris Source是有界流，不支持CDC读取。

Flink能否读取Doris并执行条件下推？

• 通过配置doris.filter.query参数可以实现。

如何写入位图类型？

CREATE TABLE bitmap_sink (
dt int,
page string,
user_id int
)
WITH (
   'connector' = 'doris',
   'fenodes' = '127.0.0.1:8030',
   'table.identifier' = 'test.bitmap_test',
   'username' = 'root',
   'password' = '',
   'sink.label-prefix' = 'doris_label',
   'sink.properties.columns' = 'dt,page,user_id,user_id=to_bitmap(user_id)'
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

errCode = 2, detailMessage = Label [label_0_1] has already been used, relate to txn [19650]

• 在Exactly-Once场景中，Flink Job必须从最新的Checkpoint/Savepoint重新启动，否则会报告上述错误。当不需要Exactly-Once时，可以通过关闭2PC提交（sink.enable-2pc=false）或更改不同的sink.label-prefix来解决。

errCode = 2, detailMessage = transaction [19650] not found

• 发生在Commit阶段，Checkpoint中记录的事务ID在FE端已过期，在此时再次提交时会出现上述错误。此时无法从Checkpoint启动，可以通过修改fe.conf中的streaming_label_keep_max_second配置来延长过期时间，默认为12小时。

errCode = 2, detailMessage = current running txns on db 10006 is 100, larger than limit 100

• 这是因为同一库的并发导入超过了100，可以通过调整fe.conf的max_running_txn_num_per_db参数来解决。详细信息，请参考max_running_txn_num_per_db。
• 同时，如果一个任务频繁修改标签并重新启动，也可能导致此错误发生。在2pc场景（重复/聚合模型）中，每个任务的标签需要唯一，在从Checkpoint重新启动时，Flink任务将主动中止之前已经成功预提交但未提交的事务。频繁修改标签并重新启动将导致大量已成功预提交的事务无法中止，占用事务。在Unique模型下，也可以关闭2pc，实现幂等写入。

当Flink向Uniq模型写入一批数据时，如何确保数据的顺序？

• 您可以添加序列列的配置来确保顺序。

Flink任务没有报错，但数据无法同步？

• 在Connector1.1.0之前，数据是批量写入的，并且写入是由数据驱动的。需要确定上游是否有数据写入。在1.1.0之后，它依赖于Checkpoint，并且必须启用Checkpoint才能进行写入。

tablet writer write failed, tablet_id=190958, txn_id=3505530, err=-235

• 通常发生在Connector1.1.0之前，这是因为写入频率过快，导致版本过多。可以通过设置sink.batch.size和sink.batch.interval参数来减少Streamload的频率。

源表和Doris表应如何对应？

• 在使用Flink Connector导入数据时，需要注意两个方面。第一，源表的列和类型应与Flink SQL中的列和类型对应；第二，Flink SQL中的列和类型必须与Doris表的列和类型匹配。

TApplicationException: get_next failed: out of sequence response: expected 4 but got 3

• 这是由于 Thrift 中的并发错误造成的。建议您尽可能使用最新的连接器和兼容的 Flink 版本。

DorisRuntimeException: Fail to abort transaction 26153 with url http://192.168.0.1:8040/api/table_name/_stream_load_2pc

• 您可以在TaskManager中搜索日志中止事务响应，并根据HTTP返回码判断是客户端问题还是服务器问题。

org.apache.flink.table.api.SqlParserException when using doris.filter.query: SQL parsing failed. “xx” encountered at row x, column xx

• 这个问题主要是由于条件中的varchar/string类型，需要进行引号转义。正确的写法是xxx = ‘‘xxx’’。这样，Flink SQL解析器会将连续的两个单引号解释为一个单引号字符，而不是字符串的结束，并将拼接的字符串作为属性的值。例如：t1 ‍>= ‘2024-01-01’ 可以写为 ‘doris.filter.query’ = ‘t1 ‍>=’‘2024-01-01’'。

Failed to connect to backend: http://host:webserver_port, and BE is still alive

• 这个问题可能是由于配置了be的IP地址，而该地址无法被外部的Flink集群访问。这主要是因为在连接fe时，be的地址是通过fe进行解析的。例如，如果将be地址添加为’127.0.0.1’，那么Flink集群通过fe获取到的be地址将是’127.0.0.1:webserver_port’，并且Flink将连接到该地址。当出现这种问题时，可以通过将be的实际对应的外部IP地址添加到"with"属性中来解决：‘benodes’=“be_ip:webserver_port,be_ip:webserver_port…”。对于整个数据库的同步，可以使用以下属性：–sink-conf benodes=be_ip:webserver,be_ip:webserver…。

当使用Flink-connector将MySQL数据同步到Doris时，时间戳之间存在几小时的时间差。

• Flink Connector默认使用UTC+8时区从MySQL同步整个数据库。如果您的数据位于不同的时区，您可以使用以下配置进行调整，例如：–mysql-conf debezium.date.format.timestamp.zone=“UTC+3”。