E-MapReduce：計算群組使用指南

查看所有Warehouse

• 文法

  SHOW WAREHOUSES [LIKE '<pattern>'];

  本文文法中的[LIKE '<pattern>']為可選內容，用於根據名稱匹配特定的Warehouse。

  • <pattern> 支援萬用字元：

    • %：匹配任意長度的字元（包括零個字元）。

    • _：匹配單個字元。

  • <pattern>樣本

    • LIKE 'test%'：匹配以 test 開頭的所有Warehouse。

    • LIKE '%_warehouse'：匹配以 _warehouse 結尾的所有Warehouse。

• 樣本

  • 查看所有Warehouse。

    SHOW WAREHOUSES;

  • 按名稱模式過濾。

    SHOW WAREHOUSES LIKE 'test%';

查看所有計算節點

每個節點的資訊中包含其所屬的Warehouse名稱。

SHOW COMPUTE NODES;

查看指定Warehouse中的計算節點

• 查看單個Warehouse的計算節點資訊

  SHOW NODES FROM WAREHOUSE <warehouse_name>; 

• 查看多個Warehouse的計算節點資訊

  SHOW NODES FROM WAREHOUSES [LIKE '<pattern>']; 

  如果不指定[LIKE '<pattern>']，則會顯示所有Warehouse的節點資訊。

查詢操作（Query）

Query僅支援通過SET_VAR方式或在當前Session中手動切換Warehouse，以指定使用特定的Warehouse。

SELECT /*+SET_VAR(warehouse="<warehouse_name>")*/ * FROM <database_name>.<table_name>;

資料匯入操作（Load）

• Stream Load

  將Warehouse資訊添加到HTTP請求的Header中。文法如下所示。

  curl --location-trusted -u <username>:<password> \
      -H "label:<label_name>" \
      -H "timeout:100" \
      -H "max_filter_ratio:1" \
      -H "warehouse:<warehouse_name>" \
      -T <file_path> \
      -XPUT http://<fe_host>:<fe_http_port>/api/<database_name>/<table_name>/_stream_load

  本文所涉及的參數說明詳見下表。

  參數名稱	是否必須	參數說明
  <username>:<password>	是	用於指定StarRocks執行個體的使用者名稱和密碼。如果帳號未設定密碼，則只需傳入<username>:，如：admin:。
  label	否	用於指定匯入作業的標籤。如果未指定標籤，StarRocks將自動為匯入作業產生一個標籤。相同標籤的資料不能多次成功匯入，以此避免資料的重複匯入。
  timeout	否	用於匯入作業的逾時時間。取值範圍：1~259200。單位：秒。
  max_filter_ratio	否	用於指定匯入作業的最大容錯率，即匯入作業能夠容忍的因資料品質不合格而過濾掉的資料行所佔的最大比例。取值範圍：0~1。
  warehouse	是	用於指定Stream Load使用的Warehouse。
  <file_path>	是	用於指定資料檔案所在的路徑。
  fe_host	是	用於指定StarRocks執行個體中FE的內網地址。該地址可在StarRocks執行個體的執行個體詳情頁面中查看。 說明 如果您直接提交匯入作業給某一個BE（或CN）節點，則需要傳入該BE（或CN）的IP地址。例如，您可以通過命令SHOW COMPUTE NODES;查看IP地址。
  fe_http_port	是	指定StarRocks執行個體中FE的HTTP連接埠號碼。預設連接埠號碼為8030。 說明 如果您直接向某一指定的BE（或CN）節點提交匯入作業，則需提供該BE（或CN）的HTTP連接埠號碼。預設連接埠號碼為8040。
  database_name	是	指定目標StarRocks表所屬資料庫的名稱。
  table_name	是	指定目標StarRocks表的名稱。

• 外部事務匯入

  在調用/api/transaction/begin介面開啟事務時，您必須指定標籤Label，並將Warehouse資訊添加到HTTP請求的Header中。文法如下所示。

  curl --location-trusted -u <username>:<password> \
      -H "label:<label_name>" \
      -H "db:<database_name>" \
      -H "table:<table_name>" \
      -H "warehouse::<warehouse_name>" \
      -XPOST http://<fe_host>:<fe_http_port>/api/transaction/begin

  之後，在使用/api/transaction/load、/api/transaction/prepare、/api/transaction/commit介面時，必須使用與/api/transaction/begin介面相同的標籤，但不需要在HTTP請求的Header中再次指定warehouse資訊。

• Broker Load

  LOAD LABEL [<database_name>.]<label_name>
  (
      data_desc[, data_desc ...]
  )
  WITH BROKER
  (
      StorageCredentialParams
  )
  [PROPERTIES
  (
      opt_properties
  )
  ]

  建立匯入任務的語句中opt_properties新增 "warehouse" 參數，預設值為當前Session的Warehouse。例如：

  LOAD LABEL my_db.path_extra_fields_label
  (
    DATA INFILE( "hdfs://<hdfs_ip>:9000/starrocks_test_data/hdfs_data/*.orc")
    INTO TABLE `my_tbl` 
    FORMAT AS "orc" 
  )  
  WITH BROKER hdfs_broker 
  (
    "hadoop.security.authentication" = "simple",
    "username" = "<hdfs_username>",
    "password" = "<hdfs_password>"
  ) 
   PROPERTIES
  (
    "warehouse"="test"
  );

• Routine load

  job_properties新增warehouse參數，預設值為當前Session的Warehouse。

  CREATE ROUTINE LOAD <database_name>.<job_name> ON <table_name>
  [load_properties]
  [job_properties]
  FROM data_source
  [data_source_properties]

  使用Routine Load指定名為test的Warehouse，通過持續消費Kafka中的訊息並匯入至StarRocks中，樣本如下。

  CREATE ROUTINE LOAD my_db.routine_load_basic_types_fe824e00_680a_11ee_81d7_00163e135aa9 
  ON my_table
  COLUMNS (k1,k2,k3,k4,k5,k6,v1,v2,v3,v4,v5,v6,v7,v8), 
  COLUMNS TERMINATED BY ',',where v1 != 0 
  PROPERTIES (
      "desired_concurrent_number"="1",
      "max_error_number"="1000",
      "max_batch_interval"="7",
      "warehouse"="test"
  )
  FROM KAFKA (
      "kafka_broker_list"="<kafka_broker_ip>:<kafka_broker_port>",
      "kafka_topic"="xxx",
      "kafka_partitions"="0",
      "kafka_offsets"="OFFSET_BEGINNING"
  );

• Flink Connector

  StarRocks提供了對Apache Flink的原生支援，可以通過Flink將資料高效地寫入StarRocks。從Flink 1.2.9及之後版本開始，您可以通過配置sink.properties.warehouse參數，將資料寫入到指定的計算群組（Warehouse），從而實現更靈活的資源管理和效能最佳化。

  使用sink.properties.warehouse指定目標計算群組（Warehouse）。

  sink.properties.warehouse=<warehouse_name>

  • <warehouse_name>：目標計算群組的名稱。

  • 如果未指定該參數，預設會使用default_warehouse計算群組。

物化視圖（Materialized View）

CREATE MATERIALIZED VIEW語句中PROPERTIES新增warehouse參數，預設值為當前Session的Warehouse。文法如下：

CREATE MATERIALIZED VIEW [IF NOT EXISTS] [database_name.]<mv_name>
[COMMENT ""]
[PROPERTIES ("key"="value", ...)]
AS 
<query_statement>

例如：要建立一個名為order_mv的物化視圖，指定後續改寫操作都在名為test的Warehouse中運行，建立語句如下：

CREATE MATERIALIZED VIEW order_mv
DISTRIBUTED BY HASH(`order_id`)
REFRESH ASYNC EVERY (interval 1 MINUTE) 
PROPERTIES ("warehouse" = "test")
AS SELECT 
    order_list.order_id,
    sum(goods.price) as total
FROM order_list INNER JOIN goods ON goods.item_id1 = order_list.item_id2
GROUP BY order_id;

執行成功後，可以用以下兩種方式查看Warehouse資訊。fe.audit.log中也會記錄物化視圖改寫語句的執行情況，其中包含了Warehouse欄位。

• 方式一：執行以下命令，查詢名為“order_mv”的物化視圖的詳細資料。

  SHOW materialized views WHERE name="order_mv";

• 方式二：執行以下命令，查看建立物化視圖“order_mv”的完整SQL語句。

  SHOW CREATE materialized VIEW order_mv;

Compaction

Compaction預設是在default_warehouse中執行的。如果想要指定其他Warehouse，可通過修改配置項lake_compaction_warehouse實現。支援在Session中動態修改和在fe.conf中修改後重啟兩種方式。

例如，想要指定名為test的Warehouse做Compaction。

• 方式一：您可以通過ADMIN SET FRONTEND CONFIG命令線上修改FE動態參數。

  ADMIN SET FRONTEND CONFIG ("lake_compaction_warehouse" = "test");

• 方式二：在fe.conf中寫入lake_compaction_warehouse = test;後重啟FE以使修改生效。

修改後可通過show proc '/compactions' ;查看最近的Compaction記錄，最後一列是Warehouse資訊。

為使用者指定預設Warehouse

您可以通過修改使用者（例如，sr_user）屬性，或在建立使用者時指定預設使用的Warehouse（例如，sr_warehouse）。

• 方法一：通過ALTER USER修改現有使用者的屬性

  ALTER USER sr_user SET PROPERTIES ('session.warehouse' = 'sr_warehouse');

• 方法二：通過CREATE USER建立新使用者並指定預設Warehouse

  CREATE USER 'sr_user' PROPERTIES ('session.warehouse' = 'sr_warehouse');

指定後，您可以通過執行SHOW PROPERTY命令來查看特定使用者的屬性。