使用灵活列更新方式导入数据至Doris集群

适用场景

在使用CDC方式将数据实时同步至Doris时，源端通常仅输出主键及变更列（即部分更新数据）。这种情况下，同一时间窗口内不同记录的更新列可能不一致，此时就可以使用灵活列更新方式将数据导入Doris。

使用限制

• 与普通部分列更新一致，灵活列更新要求每一行数据必须包含目标表的所有Key列。缺失Key列的行将被自动过滤并计入filter rows，当filter rows超过本次导入“max_filter_ratio”设定的阈值时，整个导入任务将失败。所有被过滤的数据行均会在Error Log中记录一条日志。
• JSON对象中的键值对仅当Key与目标表的列名完全一致时才生效，不匹配的键值对将被忽略。Key为“__DORIS_VERSION_COL__”、“__DORIS_ROW_STORE_COL__”和“__DORIS_SKIP_BITMAP_COL__”的键值对将被自动忽略，无需在数据中提供。
• 不支持在包含Variant列的表上使用灵活列更新。
• 不支持在有同步物化视图的表上使用灵活列更新。
• 使用灵活列更新时，不能指定或开启以下导入参数：

  merge_type、delete、fuzzy_parse、columns、jsonpaths、hidden_columns、function_column.sequence_col、sql、memtable_on_sink_node、group_commit、where。

• 仅支持JSON格式输入，不支持CSV格式。
• 不支持Cluster Keys，数据中不能包含Cluster Keys相关的列或配置。

使用方式

开启灵活列更新功能后，即可在导入任务中启用灵活列更新。

1. 新建表。在建表时，需要指定以下属性开启Merge-on-Write实现，并使表具备灵活列更新所需的bitmap隐藏列：

   "enable_unique_key_merge_on_write" = "true"
   "enable_unique_key_skip_bitmap_column" = "true"

2. 执行以下命令，若返回结果中包含"enable_unique_key_skip_bitmap_column" = "true"表示开启灵活列更新功能成功。

   SHOW CREATE TABLE db1.tbl1

3. 灵活列导入方式。

   当前灵活列仅支持通过Stream Load方式导入数据，在导入任务的“Header”中配置以下内容：

   表1 灵活列导入方式

   导入方式	配置项	值
   Stream Load	header	unique_key_update_mode:UPDATE_FLEXIBLE_COLUMNS

示例

通过一个完整的示例演示灵活列更新的使用方法。

1. 创建测试表t1。

   create database if not exists d1;
   use d1;
   CREATE TABLE t1 (
        `k` INT NULL,
        `v1` BIGINT NULL,
        `v2` BIGINT NULL DEFAULT "9876",
        `v3` BIGINT NOT NULL,
        `v4` BIGINT NOT NULL DEFAULT "1234",
        `v5` BIGINT NULL
   ) UNIQUE KEY(`k`) 
   DISTRIBUTED BY HASH(`k`) BUCKETS 1
   PROPERTIES (
        "replication_num" = "3",
        "enable_unique_key_merge_on_write" = "true",
        "enable_unique_key_skip_bitmap_column" = "true"
   ); 

2. 查看测试表t1中已有的原始数据。

   MySQL root@127.1:d1> select * from t1; 
   +---+----+----+----+----+----+
   | k | v1 | v2 | v3 | v4 | v5 |
   +---+----+----+----+----+----+
   | 0 | 0  | 0  | 0  | 0  | 0  |
   | 1 | 1  | 1  | 1  | 1  | 1  |
   | 2 | 2  | 2  | 2  | 2  | 2  |
   | 3 | 3  | 3  | 3  | 3  | 3  |
   | 4 | 4  | 4  | 4  | 4  | 4  |
   | 5 | 5  | 5  | 5  | 5  | 5  |
   +---+----+----+----+----+----+

3. 准备灵活列更新数据。每一行仅包含主键“k”和被更新的列，各行的更新列可以不同。

   {"k": 0, "__DORIS_DELETE_SIGN__": 1}
   {"k": 1, "v1": 10}
   {"k": 2, "v2": 20, "v5": 25}
   {"k": 3, "v3": 30}
   {"k": 4, "v4": 20, "v1": 43, "v3": 99}
   {"k": 5, "v5": null}
   {"k": 6, "v1": 999, "v3": 777}
   {"k": 2, "v4": 222}
   {"k": 1, "v2": 111, "v3": 111}

4. 通过Stream Load方式导入数据。

   curl -k --location-trusted -u root: \
        -H "Expect:100-continue" \
        -H "strict_mode:false" \
        -H "format:json" \
        -H "read_json_by_line:true" \
        -H "unique_key_update_mode:UPDATE_FLEXIBLE_COLUMNS" \
        -T test1.json \
        -XPUT https://<host>:<https_port>/api/d1/t1/_stream_load 

5. 查看更新结果。

   MySQL root@127.1:d1> select * from t1; 
   +---+-----+------+-----+------+--------+
   | k | v1  | v2   | v3  | v4   | v5     |
   +---+-----+------+-----+------+--------+
   | 1 | 10  | 111  | 111 | 1    | 1      |
   | 2 | 2   | 20   | 2   | 222  | 25     |
   | 3 | 3   | 3    | 30  | 3    | 3      |
   | 4 | 43  | 4    | 99  | 20   | 4      |
   | 5 | 5   | 5    | 5   | 5    | <null> |
   | 6 | 999 | 9876 | 777 | 1234 | <null> |
   +---+-----+------+-----+------+--------+