E-MapReduce:Warehouse usage guide

Query information about all warehouses

• Syntax

  SHOW WAREHOUSES [LIKE '<pattern>'];

  [LIKE '<pattern>'] is optional and is used to match a specific warehouse by name.

  • <pattern> supports the following wildcards:

    • %: matches characters of any length, including zero characters.

    • _: matches a single character.

  • Examples of <pattern>:

    • LIKE 'test%': matches all warehouses whose names start with test.

    • LIKE '%_warehouse': matches all warehouses whose names end with _warehouse.

• Examples

  • Query information about all warehouses.

    SHOW WAREHOUSES;

  • Query information about warehouses whose names match the specified pattern.

    SHOW WAREHOUSES LIKE 'test%';

Query information about all nodes in warehouses

The information of each node includes the name of the warehouse to which it belongs.

SHOW COMPUTE NODES;

Query information about nodes in a specific warehouse

• Query information about nodes in a single warehouse.

  SHOW NODES FROM WAREHOUSE <warehouse_name>; 

• Query information about nodes in multiple warehouses.

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

  If you do not specify [LIKE '<pattern>'], information about nodes in all warehouses is displayed.

Queries

You can specify a warehouse for a query operation only by using SET_VAR or manually switching the warehouse in the current session.

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

Data import operations (Load)

• Stream Load

  Add the warehouse information to the header of the HTTP request. The following syntax is supported:

  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

  The following table describes the parameters in the code.

  Parameter	Required	Description
  <username>:<password>	Yes	The username and password of the StarRocks instance. If no password is set for the account, you only need to enter <username>:, such as admin:.
  label	No	The label of the import job. If you do not specify a label, StarRocks automatically generates one for the import job. Data with the same label cannot be imported repeatedly. This prevents repeated import of the same data.
  timeout	No	The timeout period for the import job. Valid values: 1 to 259200. Unit: seconds.
  max_filter_ratio	No	The maximum fault tolerance rate of the import job. The maximum fault tolerance rate refers to the maximum proportion of data rows filtered out due to unqualified data quality that the import job can tolerate. Valid values: 0 to 1.
  warehouse	Yes	The warehouse that is used to perform the Stream Load operation.
  <file_path>	Yes	The path where the data file is located.
  fe_host	Yes	The internal endpoint of the frontend (FE) node in the StarRocks instance. You can view the endpoint on the Instance Details page of the StarRocks instance. Note If you submit the import job to a BE node or a CN, you must specify the IP address of the BE node or CN. For example, you can run the SHOW COMPUTE NODES; command to view the IP address.
  fe_http_port	Yes	The HTTP port number of the FE node in the StarRocks instance. Default value: 8030. Note If you submit the import job to a BE node or a CN, you must specify the HTTP port number of the BE node or CN. Default value: 8040.
  database_name	Yes	The name of the database where the destination StarRocks table resides.
  table_name	Yes	The name of the destination StarRocks table.

• External transaction import

  When you call the /api/transaction/begin interface to enable a transaction, you must specify a label and add the warehouse information to the HTTP request header. The following syntax is supported:

  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

  After you run the preceding command, when you call the /api/transaction/load, /api/transaction/prepare, or /api/transaction/commit interface, you must use the label that is specified for the /api/transaction/begin interface, but you do not need to specify warehouse information again in the HTTP request header.

• Broker Load

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

  Add the warehouse parameter to opt_properties in the statement that is used to create an import task. The warehouse of the current session is used by default. Sample statement:

  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

  Add the warehouse parameter to job_properties in the statement. The warehouse of the current session is used by default.

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

  Specify a warehouse named test for Routine Load to continuously consume Kafka messages and import the messages to StarRocks. Sample statement:

  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 supports native integration with Apache Flink, allowing efficient data ingestion via Flink into StarRocks. Starting from Apache Flink version 1.2.9, you can configure the sink.properties.warehouse parameter to write data into a specified warehouse, enabling more flexible resource management and performance optimization.

  Use sink.properties.warehouse to specify the target warehouse:

  sink.properties.warehouse=<warehouse_name>

  • <warehouse_name>: The name of the target warehouse.

  • If this parameter is not specified, the system will use default_warehouse.

Materialized views

Add the warehouse parameter to PROPERTIES in the CREATE MATERIALIZED VIEW statement. The warehouse of the current session is used by default. The following syntax is supported:

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

For example, create a materialized view named order_mv and specify a warehouse named test for supporting subsequent rewrite operations on the materialized view. Sample statement:

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;

After the execution is successful, you can use one of the following methods to view the information about the warehouse. The fe.audit.log file also records the execution status of rewrite statements on the materialized view, which contains the Warehouse field.

• Method 1: Run the following command to view the details of the order_mv materialized view:

  SHOW materialized views WHERE name="order_mv";

• Method 2: Run the following command to view the complete SQL statement for creating the order_mv materialized view:

  SHOW CREATE materialized VIEW order_mv;

Compaction

Compaction is executed in default_warehouse by default. If you want to specify a different warehouse, you can modify the lake_compaction_warehouse configuration item. You can dynamically modify this configuration item in sessions. You can also modify this configuration item in the fe.conf file and then restart the FE node.

For example, you want to specify a warehouse named test for compaction.

• Method 1: You can run the ADMIN SET FRONTEND CONFIG command to dynamically modify the configuration item.

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

• Method 2: Add lake_compaction_warehouse = test; to the fe.conf file and restart the FE node to make the modification take effect.

After the modification, you can run the show proc '/compactions' ; command to view the latest compaction records. The warehouse information is in the last column.