StarRocks大批量数据导入方案-使用 Kafka connector 导入数据
本文详细介绍如何使用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;