E-MapReduce：Routine Load

環境要求

• 支援訪問無認證或使用SSL方式認證的Kafka叢集。

• 支援的訊息格式如下：

  • CSV文字格式設定，每一個message為一行，且行尾不包含分行符號。

  • JSON文字格式設定。

• 不支援Array類型。

• 僅支援Kafka 0.10.0.0及以上版本。

建立匯入任務

• 文法

  CREATE ROUTINE LOAD <database>.<job_name> ON <table_name>
      [COLUMNS TERMINATED BY "column_separator" ,]
      [COLUMNS (col1, col2, ...) ,]
      [WHERE where_condition ,]
      [PARTITION (part1, part2, ...)]
      [PROPERTIES ("key" = "value", ...)]
      FROM [DATA_SOURCE]
      [(data_source_properties1 = 'value1',
      data_source_properties2 = 'value2',
      ...)]

  相關參數描述如下表所示。

  參數	是否必填	描述
  job_name	是	匯入任務的名稱，首碼可以攜帶匯入資料庫名稱，常見命名方式為時間戳記+表名。 一個DataBase內，任務名稱不可重複。
  table_name	是	匯入的目標表的名稱。
  COLUMNS TERMINATED子句	否	指定來源資料檔案中的資料行分隔符號，分隔字元預設為\t。
  COLUMNS子句	否	指定來源資料中列和表中列的映射關係。 映射列：例如，目標表有三列col1、col2和col3，來源資料有4列，其中第1、2、4列分別對應col2、col1和col3，則書寫為COLUMNS (col2, col1, temp, col3)，其中temp列為不存在的一列，用於跳過來源資料中的第三列。 衍生列：除了直接讀取來源資料的列內容之外，StarRocks還提供對資料列的加工操作。例如，目標表後加入了第四列col4，其結果由col1 + col2產生，則可以書寫為COLUMNS (col2, col1, temp, col3, col4 = col1 + col2)。
  WHERE子句	否	指定過濾條件，可以過濾掉不需要的行。過濾條件可以指定映射列或衍生列。 例如，只匯入k1大於100並且k2等於1000的行，則書寫為WHERE k1 > 100 and k2 = 1000。
  PARTITION子句	否	指定匯入目標表的Partition。如果不指定，則會自動匯入到對應的Partition中。
  PROPERTIES子句	否	指定匯入任務的通用參數。
  desired_concurrent_number	否	匯入並發度，指定一個匯入任務最多會被分成多少個子任務執行。必須大於0，預設值為3。
  max_batch_interval	否	每個子任務的最大執行時間。範圍為5~60，單位是秒。預設值為10。 1.15版本後，該參數表示子任務的調度時間，即任務多久執行一次。任務的消費資料時間為fe.conf中的routine_load_task_consume_second，預設為3s。任務的執行逾時時間為fe.conf中的routine_load_task_timeout_second，預設為15s。
  max_batch_rows	否	每個子任務最多讀取的行數。必須大於等於200000。預設值為200000。 1.15版本後，該參數只用於定義錯誤偵測視窗範圍，視窗的範圍是10 * max-batch-rows。
  max_batch_size	否	每個子任務最多讀取的位元組數。單位為位元組，範圍是100 MB到1 GB。預設值為100 MB。 1.15版本後，廢棄該參數，任務消費資料的時間為fe.conf中的routine_load_task_consume_second，預設為3s。
  max_error_number	否	採樣視窗內，允許的最大錯誤行數。必須大於等於0。預設是0，即不允許有錯誤行。 重要 被WHERE條件過濾掉的行不算錯誤行。
  strict_mode	否	是否開啟strict 模式，預設為開啟。 如果開啟後，非空未經處理資料的列類型變換為NULL，則會被過濾。關閉方式為設定該參數為false。
  timezone	否	指定匯入任務所使用的時區。 預設為使用Session的timezone參數。該參數會影響所有匯入涉及的和時區有關的函數結果。
  DATA_SOURCE	是	指定資料來源，請使用KAFKA。
  data_source_properties	否	指定資料來源相關的資訊。包括以下參數： kafka_broker_list：Kafka的Broker串連資訊，格式為ip:host。多個Broker之間以逗號（,）分隔。 kafka_topic：指定待訂閱的Kafka的Topic。 說明 如果指定資料來源相關的資訊，則kafka_broker_list和kafka_topic必填。 kafka_partitions和kafka_offsets：指定需要訂閱的Kafka Partition，以及對應的每個Partition的起始offset。 property：Kafka相關的屬性，功能等同於Kafka Shell中"--property"參數。建立匯入任務更詳細的文法可以通過執行HELP ROUTINE LOAD; 命令查看。

• 樣本：向StarRocks提交一個無認證的Routine Load匯入任務example_tbl2_ordertest，持續消費Kafka叢集中Topic ordertest2的訊息，並匯入至example_tbl2表中，匯入任務會從此Topic所指定分區的最早位點開始消費。

  CREATE ROUTINE LOAD load_test.example_tbl2_ordertest ON example_tbl
  COLUMNS(commodity_id, customer_name, country, pay_time, price, pay_dt=from_unixtime(pay_time, '%Y%m%d'))
  PROPERTIES
  (
      "desired_concurrent_number"="5",
      "format" ="json",
      "jsonpaths" ="[\"$.commodity_id\",\"$.customer_name\",\"$.country\",\"$.pay_time\",\"$.price\"]"
   )
  FROM KAFKA
  (
      "kafka_broker_list" ="<kafka_broker1_ip>:<kafka_broker1_port>,<kafka_broker2_ip>:<kafka_broker2_port>",
      "kafka_topic" = "ordertest2",
      "kafka_partitions" ="0,1,2,3,4",
      "property.kafka_default_offsets" = "OFFSET_BEGINNING"
  );

• 樣本：使用SSL安全性通訊協定訪問Kafka，具體的配置樣本如下。

  -- 指定安全性通訊協定為SSL。
  "property.security.protocol" = "ssl", 
  
   -- CA認證的位置。
  "property.ssl.ca.location" = "FILE:ca-cert",
  
  -- 如果Kafka Server端開啟了Client認證，則還需設定以下三個參數：
  -- Client的公開金鑰位置。
  "property.ssl.certificate.location" = "FILE:client.pem", 
  -- Client的私密金鑰位置。
  "property.ssl.key.location" = "FILE:client.key", 
  -- Client私密金鑰的密碼。
  "property.ssl.key.password" = "******"

  關於建立檔案的詳細資料，請參見CREATE FILE。

  說明

  在使用CREATE FILE時，請使用OSS的HTTP訪問地址作為url。具體的使用方式，請參見通過IPv6協議訪問OSS。

查看任務狀態

• 顯示load_test下，所有的例行匯入任務（包括已停止或取消的任務）。結果為一行或多行。

  USE load_test;
  SHOW ALL ROUTINE LOAD;

• 顯示load_test下，名稱為example_tbl2_ordertest的當前正在啟動並執行例行匯入任務。

  SHOW ROUTINE LOAD FOR load_test.example_tbl2_ordertest;

• 在EMR StarRocks Manager控制台，單擊中繼資料管理，單擊待查看的資料庫名稱，單擊任務，即可在Kafka匯入頁簽查看任務的執行狀態。

*************************** 1. row ***************************

                  Id: 14093
                Name: routine_load_wikipedia
          CreateTime: 2020-05-16 16:00:48
           PauseTime: N/A
             EndTime: N/A
              DbName: default_cluster:load_test
           TableName: routine_wiki_edit
               State: RUNNING
      DataSourceType: KAFKA
      CurrentTaskNum: 1
       JobProperties: {"partitions":"*","columnToColumnExpr":"event_time,channel,user,is_anonymous,is_minor,is_new,is_robot,is_unpatrolled,delta,added,deleted","maxBatchIntervalS":"10","whereExpr":"*","maxBatchSizeBytes":"104857600","columnSeparator":"','","maxErrorNum":"1000","currentTaskConcurrentNum":"1","maxBatchRows":"200000"}
DataSourceProperties: {"topic":"starrocks-load","currentKafkaPartitions":"0","brokerList":"localhost:9092"}
    CustomProperties: {}
           Statistic: {"receivedBytes":150821770,"errorRows":122,"committedTaskNum":12,"loadedRows":2399878,"loadRowsRate":199000,"abortedTaskNum":1,"totalRows":2400000,"unselectedRows":0,"receivedBytesRate":12523000,"taskExecuteTimeMs":12043}
            Progress: {"0":"13634667"}
ReasonOfStateChanged:
        ErrorLogUrls: http://172.26.**.**:9122/api/_load_error_log?file=__shard_53/error_log_insert_stmt_47e8a1d107ed4932-8f1ddf7b01ad2fee_47e8a1d107ed4932_8f1ddf7b01ad2fee, http://172.26.**.**:9122/api/_load_error_log?file=__shard_54/error_log_insert_stmt_e0c0c6b040c044fd-a162b16f6bad53e6_e0c0c6b040c044fd_a162b16f6bad53e6, http://172.26.**.**:9122/api/_load_error_log?file=__shard_55/error_log_insert_stmt_ce4c95f0c72440ef-a442bb300bd743c8_ce4c95f0c72440ef_a442bb300bd743c8
            OtherMsg:
1 row in set (0.00 sec)

暫停匯入任務

使用PAUSE語句後，此時匯入任務進入PAUSED狀態，資料暫停匯入，但任務未終止，可以通過RESUME語句重啟任務。

PAUSE ROUTINE LOAD FOR <job_name>;

暫停匯入任務後，任務的State變更為PAUSED，Statistic和Progress中的匯入資訊停止更新。此時，任務並未終止，通過SHOW ROUTINE LOAD語句可以看到已經暫停匯入任務。

恢複匯入任務

使用RESUME語句後，任務會短暫的進入NEED_SCHEDULE狀態，表示任務正在重新調度，一段時間後會重新恢複至RUNNING狀態，繼續匯入資料。

RESUME ROUTINE LOAD FOR <job_name>;

停止匯入任務

使用STOP語句讓匯入任務進入STOP狀態，資料停止匯入，任務終止，無法恢複資料匯入。

STOP ROUTINE LOAD FOR <job_name>;