本文詳細介紹如何使用Routine Load 導(dǎo)入數(shù)據(jù)

一、準備工作

1.1 安裝基礎(chǔ)環(huán)境

主要是安裝StarRocks和Kafka，本文直接跳過不做詳細介紹~

二、概念及原理

2.1 概念

• 導(dǎo)入作業(yè)（Load job）

  導(dǎo)入作業(yè)會常駐運行，當導(dǎo)入作業(yè)的狀態(tài)為 RUNNING 時，會持續(xù)不斷生成一個或多個并行的導(dǎo)入任務(wù)，不斷消費 Kafka 集群中一個 Topic 的消息，并導(dǎo)入至 StarRocks 中。

• 導(dǎo)入任務(wù)（Load task）

  導(dǎo)入作業(yè)會按照一定規(guī)則拆分成若干個導(dǎo)入任務(wù)。導(dǎo)入任務(wù)是執(zhí)行導(dǎo)入的基本單位，作為一個獨立的事務(wù)，通過 Stream Load 導(dǎo)入機制實現(xiàn)。若干個導(dǎo)入任務(wù)并行消費一個 Topic 中不同分區(qū)的消息，并導(dǎo)入至 StarRocks 中。
  

三、創(chuàng)建導(dǎo)入任務(wù)

3.1 語法

CREATE ROUTINE LOAD example_db.example_tbl1_ordertest1 ON example_tbl1
COLUMNS TERMINATED BY ",", -- 指定 CSV 數(shù)據(jù)的列分隔符，默認為\t 可不止指定
COLUMNS (order_id, pay_dt, customer_name, nationality, temp_gender, price) -- 可不指定
PROPERTIES
("desired_concurrent_number" = "5" ,-- 單個 Routine Load 導(dǎo)入作業(yè)的期望任務(wù)并發(fā)度，表示期望一個導(dǎo)入作業(yè)最多被分成多少個任務(wù)并行執(zhí)行。默認值為 3"max_batch_interval" = "5", -- 任務(wù)的調(diào)度間隔，即任務(wù)多久執(zhí)行一次。單位：秒。取值范圍：5～60。默認值：10。"format"="json", -- 源數(shù)據(jù)的格式，取值范圍：CSV、JSON。默認值：CSV"strip_outer_array" ="true" -- 是否裁剪 JSON 數(shù)據(jù)最外層的數(shù)組結(jié)構(gòu)。取值范圍：TRUE 或者 FALSE。默認值：FALSE
)
FROM KAFKA
-- 項目是kafka相關(guān)配置
("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 其他重要語法

• 查看導(dǎo)入作業(yè)：SHOW ROUTINE LOAD
• 查看導(dǎo)入任務(wù)：SHOW ROUTINE LOAD TASK

SHOW ROUTINE LOAD TASK WHERE JobName = "example_tbl2_ordertest2"

• 暫停導(dǎo)入作業(yè)：PAUSE ROUTINE LOAD

PAUSE ROUTINE LOAD FOR example_tbl2_ordertest2;

• 恢復(fù)導(dǎo)入作業(yè)：RESUME ROUTINE LOAD

RESUME ROUTINE LOAD FOR example_tbl2_ordertest2;