本文详细介绍如何使用Routine Load 导入数据

一、准备工作

1.1 安装基础环境

主要是安装StarRocks和Kafka，本文直接跳过不做详细介绍~

二、概念及原理

2.1 概念

• 导入作业（Load job）

  导入作业会常驻运行，当导入作业的状态为 RUNNING 时，会持续不断生成一个或多个并行的导入任务，不断消费 Kafka 集群中一个 Topic 的消息，并导入至 StarRocks 中。

• 导入任务（Load task）

  导入作业会按照一定规则拆分成若干个导入任务。导入任务是执行导入的基本单位，作为一个独立的事务，通过 Stream Load 导入机制实现。若干个导入任务并行消费一个 Topic 中不同分区的消息，并导入至 StarRocks 中。
  

三、创建导入任务

3.1 语法

CREATE ROUTINE LOAD example_db.example_tbl1_ordertest1 ON example_tbl1
COLUMNS TERMINATED BY ",", -- 指定 CSV 数据的列分隔符，默认为\t 可不止指定
COLUMNS (order_id, pay_dt, customer_name, nationality, temp_gender, price) -- 可不指定
PROPERTIES
(
    "desired_concurrent_number" = "5" ,-- 单个 Routine Load 导入作业的期望任务并发度，表示期望一个导入作业最多被分成多少个任务并行执行。默认值为 3
    "max_batch_interval" = "5", -- 任务的调度间隔，即任务多久执行一次。单位：秒。取值范围：5～60。默认值：10。
    "format"="json", -- 源数据的格式，取值范围：CSV、JSON。默认值：CSV
    "strip_outer_array" ="true" -- 是否裁剪 JSON 数据最外层的数组结构。取值范围：TRUE 或者 FALSE。默认值：FALSE
)
FROM KAFKA
-- 项目是kafka相关配置
(
    "kafka_broker_list" = "<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
    "kafka_topic" = "ordertest1",
    "kafka_partitions" = "0,1,2,3,4",
    "property.kafka_default_offsets" = "OFFSET_BEGINNING"
);

3.2 其他重要语法

• 查看导入作业：SHOW ROUTINE LOAD
• 查看导入任务：SHOW ROUTINE LOAD TASK

SHOW ROUTINE LOAD TASK WHERE JobName = "example_tbl2_ordertest2"

• 暂停导入作业：PAUSE ROUTINE LOAD

PAUSE ROUTINE LOAD FOR example_tbl2_ordertest2;

• 恢复导入作业：RESUME ROUTINE LOAD

RESUME ROUTINE LOAD FOR example_tbl2_ordertest2;