> For the complete documentation index, see [llms.txt](https://docs.starrocks.io/llms.txt). This page is also available as Markdown at its `.md` URL. # Load data using Routine Load tip Try Routine Load out in this [Quick Start](https://docs.starrocks.io/docs/quick_start/routine-load.md) This topic introduces how to create a Routine Load job to stream Kafka messages (events) into StarRocks, and familiarizes you with some basic concepts about Routine Load. To continuously load messages of a stream into StarRocks, you can store the message stream in a Kafka topic, and create a Routine Load job to consume the messages. The Routine Load job persists in StarRocks, generates a series of load tasks to consume the messages in all or part of the partitions in the topic, and loads the messages into StarRocks. A Routine Load job supports exactly-once delivery semantics to guarantee the data loaded into StarRocks is neither lost nor duplicated. Routine Load supports data transformation at data loading and supports data changes made by UPSERT and DELETE operations during data loading. For more information, see [Transform data at loading](https://docs.starrocks.io/docs/loading/Etl_in_loading.md) and [Change data through loading](https://docs.starrocks.io/docs/loading/Load_to_Primary_Key_tables.md). You can load data into StarRocks tables only as a user who has the INSERT privilege on those StarRocks tables. If you do not have the INSERT privilege, follow the instructions provided in [GRANT](https://docs.starrocks.io/docs/sql-reference/sql-statements/account-management/GRANT.md) to grant the INSERT privilege to the user that you use to connect to your StarRocks cluster. The syntax is `GRANT INSERT ON TABLE IN DATABASE TO { ROLE | USER }`. ## Supported data formats[​](#supported-data-formats "Direct link to Supported data formats") Routine Load now supports consuming CSV, JSON, and Avro (supported since v3.0.1) formatted data from a Kafka cluster. > **NOTE** > > For CSV data, take note of the following points: > > * You can use a UTF-8 string, such as a comma (,), tab, or pipe (|), whose length does not exceed 50 bytes as a text delimiter. > * Null values are denoted by using `\N`. For example, a data file consists of three columns, and a record from that data file holds data in the first and third columns but no data in the second column. In this situation, you need to use `\N` in the second column to denote a null value. This means the record must be compiled as `a,\N,b` instead of `a,,b`. `a,,b` denotes that the second column of the record holds an empty string. ## Basic concepts[​](#basic-concepts "Direct link to Basic concepts") ![routine load](/assets/images/4.5.2-1-a6d6d07699a1f32c703d4b80921e2e44.png) ### Terminology[​](#terminology "Direct link to Terminology") * **Load job** A Routine Load job is a long-running job. As long as its status is RUNNING, a load job continuously generates one or multiple concurrent load tasks which consume the messages in a topic of a Kafka cluster and load the data into StarRocks. * **Load task** A load job is split into multiple load tasks by certain rules. A load task is the basic unit of data loading. As an individual event, a load task implements the load mechanism based on [Stream Load](https://docs.starrocks.io/docs/loading/StreamLoad.md). Multiple load tasks concurrently consume the messages from different partitions of a topic, and load the data into StarRocks. ### Workflow[​](#workflow "Direct link to Workflow") 1. **Create a Routine Load job.** To load data from Kafka, you need to create a Routine Load job by running the [CREATE ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md) statement. The FE parses the statement, and creates the job according to the properties you have specified. 2. **The FE splits the job into multiple load tasks.** The FE split the job into multiple load tasks based on certain rules. Each load task is an individual transaction. The splitting rules are as follows: * The FE calculates the actual concurrent number of the load tasks according to the desired concurrent number `desired_concurrent_number`, the partition number in the Kafka topic, and the number of the BE nodes that are alive. * The FE splits the job into load tasks based on the actual concurrent number calculated, and arranges the tasks in the task queue. Each Kafka topic consists of multiple partitions. The relation between the topic partition and the load task is as follows: * A partition is uniquely assigned to a load task, and all messages from the partition are consumed by the load task. * A load task can consume messages from one or more partitions. * All partitions are distributed evenly among load tasks. 3. **Multiple load tasks run concurrently to consume the messages from multiple Kafka topic partitions, and load the data into StarRocks** 1. **The FE schedules and submits load tasks**: the FE schedules the load tasks in the queue on a timely basis, and assigns them to selected Coordinator BE nodes. The interval between load tasks is defined by the configuration item `max_batch_interval`. The FE distributes the load tasks evenly to all BE nodes. See [CREATE ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md#examples) for more information about `max_batch_interval`. 2. The Coordinator BE starts the load task, consumes messages in partitions, parses and filters the data. A load task lasts until the pre-defined amount of messages are consumed or the pre-defined time limit is reached. The message batch size and time limit are defined in the FE configurations `max_routine_load_batch_size` and `routine_load_task_consume_second`. For detailed information, see [FE Configuration](https://docs.starrocks.io/docs/administration/configuration/FE_parameters.md). The Coordinator BE then distributes the messages to the Executor BEs. The Executor BEs write the messages to disks. > **NOTE** > > StarRocks supports access to Kafka via security protocols including SASL_SSL, SAS_PLAINTEXT, SSL, and PLAINTEXT. This topic uses connecting to Kafka via PLAINTEXT as an example. If you need to connect to Kafka via other security protocols, see [CREATE ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md). 4. **The FE generates new load tasks to load data continuously.** After the Executor BEs has written the data to disks, the Coordinator BE reports the result of the load task to the FE. Based on the result, the FE then generates new load tasks to load the data continuously. Or the FE retries the failed tasks to make sure the data loaded into StarRocks is neither lost nor duplicated. ## Create a Routine Load job[​](#create-a-routine-load-job "Direct link to Create a Routine Load job") The following three examples describe how to consume CSV-format, JSON-format and Avro-format data in Kafka, and load the data into StarRocks by creating a Routine Load job. For detailed syntax and parameter descriptions, see [CREATE ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md). ### Load CSV-format data[​](#load-csv-format-data "Direct link to Load CSV-format data") This section describes how to create a Routine Load job to consume CSV-format data in a Kafka cluster, and load the data into StarRocks. #### Prepare a dataset[​](#prepare-a-dataset "Direct link to Prepare a dataset") Suppose there is a CSV-format dataset in the topic `ordertest1` in a Kafka cluster. Every message in the dataset includes six fields: order ID, payment date, customer name, nationality, gender, and price. ```plain 2020050802,2020-05-08,Johann Georg Faust,Deutschland,male,895 2020050802,2020-05-08,Julien Sorel,France,male,893 2020050803,2020-05-08,Dorian Grey,UK,male,1262 2020050901,2020-05-09,Anna Karenina",Russia,female,175 2020051001,2020-05-10,Tess Durbeyfield,US,female,986 2020051101,2020-05-11,Edogawa Conan,japan,male,8924 ``` #### Create a table[​](#create-a-table "Direct link to Create a table") According to the fields of CSV-format data, create the table `example_tbl1` in the database `example_db`. The following example creates a table with 5 fields excluding the field of customer gender in the CSV-format data. ```sql CREATE TABLE example_db.example_tbl1 ( `order_id` bigint NOT NULL COMMENT "Order ID", `pay_dt` date NOT NULL COMMENT "Payment date", `customer_name` varchar(26) NULL COMMENT "Customer name", `nationality` varchar(26) NULL COMMENT "Nationality", `price`double NULL COMMENT "Price" ) ENGINE=OLAP DUPLICATE KEY (order_id,pay_dt) DISTRIBUTED BY HASH(`order_id`); ``` > **NOTICE** > > Since v2.5.7, StarRocks can automatically set the number of buckets (BUCKETS) when you create a table or add a partition. You no longer need to manually set the number of buckets. For detailed information, see [set the number of buckets](https://docs.starrocks.io/docs/table_design/data_distribution.md#set-the-number-of-buckets). #### Submit a Routine Load job[​](#submit-a-routine-load-job "Direct link to Submit a Routine Load job") Execute the following statement to submit a Routine Load job named `example_tbl1_ordertest1` to consume the messages in the topic `ordertest1` and load the data into the table `example_tbl1`. The load task consumes the messages from the initial offset in the specified partitions of the topic. ```sql CREATE ROUTINE LOAD example_db.example_tbl1_ordertest1 ON example_tbl1 COLUMNS TERMINATED BY ",", COLUMNS (order_id, pay_dt, customer_name, nationality, temp_gender, price) PROPERTIES ( "desired_concurrent_number" = "5" ) FROM KAFKA ( "kafka_broker_list" = ":,:", "kafka_topic" = "ordertest1", "kafka_partitions" = "0,1,2,3,4", "property.kafka_default_offsets" = "OFFSET_BEGINNING" ); ``` After submitting the load job, you can execute the [SHOW ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/SHOW_ROUTINE_LOAD.md) statement to check the status of the load job. * **load job name** There could be multiple load job on a table. Therefore, we recommend you name a load job with the corresponding Kafka topic and the time when the load job is submitted. It helps you distinguish the load job on each table. * **Column separator** The property `COLUMN TERMINATED BY` defines the column separator of the CSV-format data. The default is `\t`. * **Kafka topic partition and offset** You can specify the properties `kafka_partitions` and `kafka_offsets` to specify the partitions and offsets to consume the messages. For example, if you want the load job to consume messages from the Kafka partitions `"0,1,2,3,4"` of the topic `ordertest1` all with the initial offsets, you can specify the properties as follows: If you want the load job to consume messages from the Kafka partitions `"0,1,2,3,4"`and you need to specify a separate starting offset for each partition, you can configure as follows: ```sql "kafka_partitions" ="0,1,2,3,4", "kafka_offsets" = "OFFSET_BEGINNING, OFFSET_END, 1000, 2000, 3000" ``` You can also set the default offsets of all partitions with the property `property.kafka_default_offsets`. ```sql "kafka_partitions" ="0,1,2,3,4", "property.kafka_default_offsets" = "OFFSET_BEGINNING" ``` For detailed information, see [CREATE ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md). * **Data mapping and transformation** To specify the mapping and transformation relationship between the CSV-format data, and the StarRocks table, you need to use the `COLUMNS` parameter. **Data mapping:** * StarRocks extracts the columns in the CSV-format data and maps them **in sequence** onto the fields declared in the `COLUMNS` parameter. * StarRocks extracts the fields declared in the `COLUMNS` parameter and maps them **by name** onto the columns of StarRocks table. **Data transformation:** And because the example excludes the column of customer gender from the CSV-format data, the field `temp_gender` in `COLUMNS` parameter is used as a placeholder for this field. The other fields are mapped to columns of the StarRocks table `example_tbl1` directly. For more information about data transformation, see [Transform data at loading](https://docs.starrocks.io/docs/loading/Etl_in_loading.md). > **NOTE** > > You do not need to specify the `COLUMNS` parameter if the names, number, and order of the columns in the CSV-format data completely correspond to those of the StarRocks table. * **Task concurrency** When there are many Kafka topic partitions and enough BE nodes, you can accelerate the loading by increasing the task concurrency. To increase the actual load task concurrency, you can increase the desired load task concurrency `desired_concurrent_number` when you create a routine load job. You can also set the dynamic configuration item of FE `max_routine_load_task_concurrent_num` ( default maximum load task currency ) to a larger value. For more information about `max_routine_load_task_concurrent_num`, please see [FE configuration items](https://docs.starrocks.io/docs/administration/configuration/FE_parameters.md). The actual task concurrency is defined by the minimum value among the number of BE nodes that are alive, the number of the pre-specified Kafka topic partitions, and the values of `desired_concurrent_number` and `max_routine_load_task_concurrent_num`. In the example, the number of BE nodes that are alive is `5`, the number of the pre-specified Kafka topic partitions is `5`, and the value of `max_routine_load_task_concurrent_num` is `5`. To increase the actual load task concurrency, you can increase the `desired_concurrent_number` from the default value `3` to `5`. For more about the properties, see [CREATE ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md). ### Load JSON-format data[​](#load-json-format-data "Direct link to Load JSON-format data") This section describes how to create a Routine Load job to consume JSON-format data in a Kafka cluster, and load the data into StarRocks. #### Prepare a dataset[​](#prepare-a-dataset-1 "Direct link to Prepare a dataset") Suppose there is a JSON-format dataset in the topic `ordertest2` in a Kafka cluster. The dataset includes six keys: commodity ID, customer name, nationality, payment time, and price. Besides, you want to transform the payment time column into the DATE type, and load it into the `pay_dt` column in the StarRocks table. ```json {"commodity_id": "1", "customer_name": "Mark Twain", "country": "US","pay_time": 1589191487,"price": 875} {"commodity_id": "2", "customer_name": "Oscar Wilde", "country": "UK","pay_time": 1589191487,"price": 895} {"commodity_id": "3", "customer_name": "Antoine de Saint-Exupéry","country": "France","pay_time": 1589191487,"price": 895} ``` > **CAUTION** Each JSON object in a row must be in one Kafka message, otherwise a JSON parsing error is returned. #### Create a table[​](#create-a-table-1 "Direct link to Create a table") According to the keys of the JSON-format data, create the table `example_tbl2` in the database `example_db`. ```sql CREATE TABLE `example_tbl2` ( `commodity_id` varchar(26) NULL COMMENT "Commodity ID", `customer_name` varchar(26) NULL COMMENT "Customer name", `country` varchar(26) NULL COMMENT "Country", `pay_time` bigint(20) NULL COMMENT "Payment time", `pay_dt` date NULL COMMENT "Payment date", `price`double SUM NULL COMMENT "Price" ) ENGINE=OLAP AGGREGATE KEY(`commodity_id`,`customer_name`,`country`,`pay_time`,`pay_dt`) DISTRIBUTED BY HASH(`commodity_id`); ``` > **NOTICE** > > Since v2.5.7, StarRocks can automatically set the number of buckets (BUCKETS) when you create a table or add a partition. You no longer need to manually set the number of buckets. For detailed information, see [set the number of buckets](https://docs.starrocks.io/docs/table_design/data_distribution.md#set-the-number-of-buckets). #### Submit a Routine Load job[​](#submit-a-routine-load-job-1 "Direct link to Submit a Routine Load job") Execute the following statement to submit a Routine Load job named `example_tbl2_ordertest2` to consume the messages in the topic `ordertest2` and load the data into the table `example_tbl2`. The load task consumes the messages from the initial offset in the specified partitions of the topic. ```sql CREATE ROUTINE LOAD example_db.example_tbl2_ordertest2 ON example_tbl2 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_topic" = "ordertest2", "kafka_partitions" ="0,1,2,3,4", "property.kafka_default_offsets" = "OFFSET_BEGINNING" ); ``` After submitting the load job, you can execute the [SHOW ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/SHOW_ROUTINE_LOAD.md) statement to check the status of the load job. * **Data format** You need to specify `"format" = "json"` in the clause `PROPERTIES` to define that the data format is JSON. * **Data mapping and transformation** To specify the mapping and transformation relationship between the JSON-format data, and the StarRocks table, you need to specify the parameter `COLUMNS` and property`jsonpaths`. The order of fields specified in the `COLUMNS` parameter must match that of the JSON-format data, and the name of fields must match that of the StarRocks table. The property `jsonpaths` is used to extract the required fields from the JSON data. These fields are then named by the property `COLUMNS`. Because the example needs to transform the payment time field to the DATE data type, and load the data into the `pay_dt` column in the StarRocks table, you need to use the from_unixtime function. The other fields are mapped to fields of the table `example_tbl2` directly. **Data mapping:** * StarRocks extracts the `name` and `code` keys of JSON-format data and maps them onto the keys declared in the `jsonpaths` property. * StarRocks extracts the keys declared in the `jsonpaths` property and maps them **in sequence** onto the fields declared in the `COLUMNS` parameter. * StarRocks extracts the fields declared in the `COLUMNS` parameter and maps them **by name** onto the columns of StarRocks table. **Data transformation**: * Because the example needs to transform the key `pay_time` to the DATE data type, and load the data into the `pay_dt` column in the StarRocks table, you need to use the from_unixtime function in `COLUMNS` parameter. The other fields are mapped to fields of the table `example_tbl2` directly. * And because the example excludes the column of customer gender from the JSON-format data, the field `temp_gender` in `COLUMNS` parameter is used as a placeholder for this field. The other fields are mapped to columns of the StarRocks table `example_tbl1` directly. For more information about data transformation, see [Transform data at loading](https://docs.starrocks.io/docs/loading/Etl_in_loading.md). > **NOTE** > > You do not need to specify the `COLUMNS` parameter if the names and number of the keys in the JSON object completely match those of fields in the StarRocks table. ### Load Avro-format data[​](#load-avro-format-data "Direct link to Load Avro-format data") Since v3.0.1, StarRocks supports loading Avro data by using Routine Load. #### Prepare a dataset[​](#prepare-a-dataset-2 "Direct link to Prepare a dataset") ##### Avro schema[​](#avro-schema "Direct link to Avro schema") 1. Create the following Avro schema file `avro_schema.avsc`: ```json { "type": "record", "name": "sensor_log", "fields" : [ {"name": "id", "type": "long"}, {"name": "name", "type": "string"}, {"name": "checked", "type" : "boolean"}, {"name": "data", "type": "double"}, {"name": "sensor_type", "type": {"type": "enum", "name": "sensor_type_enum", "symbols" : ["TEMPERATURE", "HUMIDITY", "AIR-PRESSURE"]}} ] } ``` 2. Register the Avro schema in the [Schema Registry](https://docs.confluent.io/cloud/current/get-started/schema-registry.html#create-a-schema). ##### Avro data[​](#avro-data "Direct link to Avro data") Prepare the Avro data and send it to the Kafka topic `topic_0`. #### Create a table[​](#create-a-table-2 "Direct link to Create a table") According to the fields of Avro data, create a table `sensor_log` in the target database `example_db` in the StarRocks cluster. The column names of the table must match the field names in the Avro data. For the data type mapping between the table columns and the Avro data fields, see \[Data types mapping]\(#Data types mapping). ```sql CREATE TABLE example_db.sensor_log ( `id` bigint NOT NULL COMMENT "sensor id", `name` varchar(26) NOT NULL COMMENT "sensor name", `checked` boolean NOT NULL COMMENT "checked", `data` double NULL COMMENT "sensor data", `sensor_type` varchar(26) NOT NULL COMMENT "sensor type" ) ENGINE=OLAP DUPLICATE KEY (id) DISTRIBUTED BY HASH(`id`); ``` > **NOTICE** > > Since v2.5.7, StarRocks can automatically set the number of buckets (BUCKETS) when you create a table or add a partition. You no longer need to manually set the number of buckets. For detailed information, see [set the number of buckets](https://docs.starrocks.io/docs/table_design/data_distribution.md#set-the-number-of-buckets). #### Submit a Routine Load job[​](#submit-a-routine-load-job-2 "Direct link to Submit a Routine Load job") Execute the following statement to submit a Routine Load job named `sensor_log_load_job` to consume the Avro messages in the Kafka topic `topic_0` and load the data into the table `sensor_log` in the database `sensor`. The load job consumes the messages from the initial offset in the specified partitions of the topic. ```sql CREATE ROUTINE LOAD example_db.sensor_log_load_job ON sensor_log PROPERTIES ( "format" = "avro" ) FROM KAFKA ( "kafka_broker_list" = ":,:,...", "confluent.schema.registry.url" = "http://172.xx.xxx.xxx:8081", "kafka_topic" = "topic_0", "kafka_partitions" = "0,1,2,3,4,5", "property.kafka_default_offsets" = "OFFSET_BEGINNING" ); ``` * Data Format You need to specify `"format = "avro"` in the clause `PROPERTIES` to define that the data format is Avro. * Schema Registry You need to configure `confluent.schema.registry.url` to specify the URL of the Schema Registry where the Avro schema is registered. StarRocks retrieves the Avro schema by using this URL. The format is as follows: ```plaintext confluent.schema.registry.url = http[s]://[:@][:] ``` * Data mapping and transformation To specify the mapping and transformation relationship between the Avro-format data and the StarRocks table, you need to specify the parameter `COLUMNS` and property `jsonpaths`. The order of fields specified in the `COLUMNS` parameter must match that of the fields in the property `jsonpaths`, and the names of fields must match these of the StarRocks table. The property `jsonpaths` is used to extract the required fields from the Avro data. These fields are then named by the property `COLUMNS`. For more information about data transformation, see [Transform data at loading](https://docs.starrocks.io/docs/loading/Etl_in_loading.md). > NOTE > > You do not need to specify the `COLUMNS` parameter if the names and number of the fields in the Avro record completely match those of columns in the StarRocks table. After submitting the load job, you can execute the [SHOW ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/SHOW_ROUTINE_LOAD.md) statement to check the status of the load job. #### Data types mapping[​](#data-types-mapping "Direct link to Data types mapping") The data type mapping between the Avro data fields you want to load and the StarRocks table columns is as follows: ##### Primitive types[​](#primitive-types "Direct link to Primitive types") | Avro | StarRocks | | ------- | --------- | | nul | NULL | | boolean | BOOLEAN | | int | INT | | long | BIGINT | | float | FLOAT | | double | DOUBLE | | bytes | STRING | | string | STRING | ##### Complex types[​](#complex-types "Direct link to Complex types") | Avro | StarRocks | | -------------- | -------------------------------------------------------------------------- | | record | STRUCT, or load the entire RECORD or its subfields into StarRocks as JSON. | | enums | STRING | | arrays | ARRAY | | maps | MAP or JSON | | union(T, null) | NULLABLE(T) | | fixed | STRING | #### Limits[​](#limits "Direct link to Limits") * Currently, StarRocks does not support schema evolution. * Each Kafka message must only contain a single Avro data record. ### Access source message metadata[​](#access-source-message-metadata "Direct link to Access source message metadata") When loading JSON- or Avro-format data, you can populate destination columns from a message's Kafka/Pulsar metadata — topic, partition, offset, timestamp, key, and headers — instead of from the message payload. You declare the metadata in an `INCLUDE METADATA (...)` clause that binds each metadata key to an alias; the alias is an ordinary source column you reference from `COLUMNS`. This is useful for auditing (which topic/partition/offset a row came from), event-time processing (using the message timestamp), and routing on a header value. #### Syntax[​](#syntax "Direct link to Syntax") ```sql INCLUDE METADATA ( [AS ] [, [AS ] ...] ) ``` `INCLUDE METADATA` is a load property; place it among the other load properties (such as `COLUMNS` and `WHERE`), in any order, before the `PROPERTIES` and `FROM` clauses. `AS ` is optional. If it is omitted, the alias defaults to ``. The alias must be unique within the clause, and it must not collide with a payload field, a destination-table column, or a reserved column name. #### Metadata keys[​](#metadata-keys "Direct link to Metadata keys") The supported keys depend on the data source. | Source | Key | Type | Description | | ------ | ----------------- | ---------------------- | ------------------------------------------------------------------------------------------------- | | KAFKA | `TOPIC` | VARCHAR | Topic name. | | KAFKA | `PARTITION` | INT | Partition number. | | KAFKA | `OFFSET` | BIGINT | Message offset within the partition. | | KAFKA | `TIMESTAMP_MS` | BIGINT | Record timestamp in milliseconds since epoch. `NULL` when the broker reports no timestamp. | | KAFKA | `KEY` | VARCHAR | Message key as raw bytes. `NULL` when the message has no key. | | KAFKA | `HEADERS` | MAP\ | All headers as a map. On duplicate keys, the last value wins. | | PULSAR | `TOPIC` | VARCHAR | The logical topic the job consumes (a partitioned topic's `-partition-N` suffix is not included). | | PULSAR | `PARTITION` | INT | Partition index, parsed from the per-message topic name. `NULL` for a non-partitioned topic. | | PULSAR | `KEY` | VARCHAR | Partition key. `NULL` when the message has no key. | | PULSAR | `MESSAGE_ID` | VARCHAR | Message ID. | | PULSAR | `PUBLISH_TIME_MS` | BIGINT | Publish time in milliseconds since epoch. | | PULSAR | `EVENT_TIME_MS` | BIGINT | Event time in milliseconds since epoch. `NULL` when the producer did not set it. | | PULSAR | `PROPERTIES` | MAP\ | All properties as a map. On duplicate keys, the last value wins. | To read a single header/property value, use `element_at(, '')` over the `HEADERS`/`PROPERTIES` map: the last value wins on a duplicate key, and the result is `NULL` when the key is absent. Header/property values are raw bytes placed into VARCHAR as-is (no UTF-8 validation). #### Usage notes[​](#usage-notes "Direct link to Usage notes") * `INCLUDE METADATA` is available for `format = json` (Kafka and Pulsar) and `format = avro` (Kafka only; Pulsar Routine Load does not support Avro). It is not supported for CSV, where one message can expand into many rows and per-message metadata would be ambiguous. * A metadata alias is an ordinary source column: reference it anywhere in the `COLUMNS` expressions. If a payload field has the same name as a metadata key, specify a different alias with `AS ` to avoid ambiguity. * `OFFSET` is Kafka-only and `MESSAGE_ID` is Pulsar-only; using a key unsupported by the source raises an error that lists the keys supported for that source. * `HEADERS`/`PROPERTIES` is a `MAP\`. The source headers/properties are an ordered list that may repeat a key; duplicates collapse into the map with the last value winning (an `element_at(map, 'name')` lookup is likewise last-wins, and returns `NULL` when the key is absent). Values are raw bytes stored in VARCHAR as-is — there is no UTF-8 validation or decoding. #### Example[​](#example "Direct link to Example") Load the order payload field `order_id` together with the source topic, partition, offset, the message timestamp converted to a `DATETIME`, and a `trace-id` header: ```sql CREATE TABLE example_db.orders_with_meta ( order_id BIGINT, src_topic VARCHAR(256), src_partition INT, src_offset BIGINT, msg_time DATETIME, trace_id VARCHAR(128) ) ENGINE = OLAP DUPLICATE KEY(order_id) DISTRIBUTED BY HASH(order_id); CREATE ROUTINE LOAD example_db.orders_with_meta_job ON orders_with_meta INCLUDE METADATA ( TOPIC AS m_topic, PARTITION AS m_partition, OFFSET AS m_offset, TIMESTAMP_MS AS m_timestamp, HEADERS AS m_headers ), COLUMNS ( order_id, src_topic = m_topic, src_partition = m_partition, src_offset = m_offset, msg_time = from_unixtime(m_timestamp / 1000), trace_id = element_at(m_headers, 'trace-id') ) PROPERTIES ( "format" = "json", "jsonpaths" = "[\"$.order_id\"]" ) FROM KAFKA ( "kafka_broker_list" = ":,...", "kafka_topic" = "topic_orders", "property.kafka_default_offsets" = "OFFSET_BEGINNING" ); ``` A metadata alias may be used inside an expression (as with `from_unixtime(m_timestamp / 1000)` above); only payload columns are listed in `jsonpaths`. ## Check a load job and task[​](#check-a-load-job-and-task "Direct link to Check a load job and task") ### Check a load job[​](#check-a-load-job "Direct link to Check a load job") Execute the [SHOW ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/SHOW_ROUTINE_LOAD.md) statement to check the status of the load job `example_tbl2_ordertest2`. StarRocks returns the execution state `State`, the statistical information (including the total rows consumed and the total rows loaded) `Statistics`, and the progress of the load job `progress`. If the state of the load job is automatically changed to **PAUSED**, it is possibly because the number of error rows has exceeded the threshold. For detailed instructions on setting this threshold, see [CREATE ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/CREATE_ROUTINE_LOAD.md). You can check the files `ReasonOfStateChanged` and `ErrorLogUrls` to identify and troubleshoot the problem. Having fixed the problem, you can then execute the [RESUME ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/RESUME_ROUTINE_LOAD.md) statement to resume the **PAUSED** load job. If the state of the load job is **CANCELLED**, it is possibly because the load job encounters an exception (such as the table has been dropped). You can check the files `ReasonOfStateChanged` and `ErrorLogUrls` to identify and troubleshoot the problem. However, you cannot resume a **CANCELLED** load job. ```sql MySQL [example_db]> SHOW ROUTINE LOAD FOR example_tbl2_ordertest2 \G *************************** 1. row *************************** Id: 63013 Name: example_tbl2_ordertest2 CreateTime: 2022-08-10 17:09:00 PauseTime: NULL EndTime: NULL DbName: default_cluster:example_db TableName: example_tbl2 State: RUNNING DataSourceType: KAFKA CurrentTaskNum: 3 JobProperties: {"partitions":"*","partial_update":"false","columnToColumnExpr":"commodity_id,customer_name,country,pay_time,pay_dt=from_unixtime(`pay_time`, '%Y%m%d'),price","maxBatchIntervalS":"20","whereExpr":"*","dataFormat":"json","timezone":"Asia/Shanghai","format":"json","json_root":"","strict_mode":"false","jsonpaths":"[\"$.commodity_id\",\"$.customer_name\",\"$.country\",\"$.pay_time\",\"$.price\"]","desireTaskConcurrentNum":"3","maxErrorNum":"0","strip_outer_array":"false","currentTaskConcurrentNum":"3","maxBatchRows":"200000"} DataSourceProperties: {"topic":"ordertest2","currentKafkaPartitions":"0,1,2,3,4","brokerList":":,:"} CustomProperties: {"kafka_default_offsets":"OFFSET_BEGINNING"} Statistic: {"receivedBytes":230,"errorRows":0,"committedTaskNum":1,"loadedRows":2,"loadRowsRate":0,"abortedTaskNum":0,"totalRows":2,"unselectedRows":0,"receivedBytesRate":0,"taskExecuteTimeMs":522} Progress: {"0":"1","1":"OFFSET_ZERO","2":"OFFSET_ZERO","3":"OFFSET_ZERO","4":"OFFSET_ZERO"} ReasonOfStateChanged: ErrorLogUrls: OtherMsg: ``` > **CAUTION** > > You cannot check a load job that has stopped or has not yet started. ### Check a load task[​](#check-a-load-task "Direct link to Check a load task") Execute the [SHOW ROUTINE LOAD TASK](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/SHOW_ROUTINE_LOAD_TASK.md) statement to check the load tasks of the load job `example_tbl2_ordertest2`, such as how many tasks are currently running, the Kafka topic partitions that are consumed and the consumption progress `DataSourceProperties`, and the corresponding Coordinator BE node `BeId`. ```sql MySQL [example_db]> SHOW ROUTINE LOAD TASK WHERE JobName = "example_tbl2_ordertest2" \G *************************** 1. row *************************** TaskId: 18c3a823-d73e-4a64-b9cb-b9eced026753 TxnId: -1 TxnStatus: UNKNOWN JobId: 63013 CreateTime: 2022-08-10 17:09:05 LastScheduledTime: 2022-08-10 17:47:27 ExecuteStartTime: NULL Timeout: 60 BeId: -1 DataSourceProperties: {"1":0,"4":0} Message: there is no new data in kafka, wait for 20 seconds to schedule again *************************** 2. row *************************** TaskId: f76c97ac-26aa-4b41-8194-a8ba2063eb00 TxnId: -1 TxnStatus: UNKNOWN JobId: 63013 CreateTime: 2022-08-10 17:09:05 LastScheduledTime: 2022-08-10 17:47:26 ExecuteStartTime: NULL Timeout: 60 BeId: -1 DataSourceProperties: {"2":0} Message: there is no new data in kafka, wait for 20 seconds to schedule again *************************** 3. row *************************** TaskId: 1a327a34-99f4-4f8d-8014-3cd38db99ec6 TxnId: -1 TxnStatus: UNKNOWN JobId: 63013 CreateTime: 2022-08-10 17:09:26 LastScheduledTime: 2022-08-10 17:47:27 ExecuteStartTime: NULL Timeout: 60 BeId: -1 DataSourceProperties: {"0":2,"3":0} Message: there is no new data in kafka, wait for 20 seconds to schedule again ``` ## Pause a load job[​](#pause-a-load-job "Direct link to Pause a load job") You can execute the [PAUSE ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/PAUSE_ROUTINE_LOAD.md) statement to pause a load job. The state of the load job will be **PAUSED** after the statement is executed. However, it has not stopped. You can execute the [RESUME ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/RESUME_ROUTINE_LOAD.md) statement to resume it. You can also check its status with the [SHOW ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/SHOW_ROUTINE_LOAD.md) statement. The following example pauses the load job `example_tbl2_ordertest2`: ```sql PAUSE ROUTINE LOAD FOR example_tbl2_ordertest2; ``` ## Resume a load job[​](#resume-a-load-job "Direct link to Resume a load job") You can execute the [RESUME ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/RESUME_ROUTINE_LOAD.md) statement to resume a paused load job. The state of the load job will be **NEED_SCHEDULE** temporarily (because the load job is being re-scheduled), and then become **RUNNING**. You can check its status with the [SHOW ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/SHOW_ROUTINE_LOAD.md) statement. The following example resumes the paused load job `example_tbl2_ordertest2`: ```sql RESUME ROUTINE LOAD FOR example_tbl2_ordertest2; ``` ## Alter a load job[​](#alter-a-load-job "Direct link to Alter a load job") Before altering a load job, you must pause it with the [PAUSE ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/PAUSE_ROUTINE_LOAD.md) statement. Then you can execute the [ALTER ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/ALTER_ROUTINE_LOAD.md). After altering it, you can execute the [RESUME ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/RESUME_ROUTINE_LOAD.md) statement to resume it, and check its status with the [SHOW ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/SHOW_ROUTINE_LOAD.md) statement. Suppose the number of the BE nodes that are alive increases to `6` and the Kafka topic partitions to be consumed is `"0,1,2,3,4,5,6,7"`. If you want to increase the actual load task concurrency, you can execute the following statement to increase the number of desired task concurrency `desired_concurrent_number` to `6` (greater than or equal to the number of BE nodes that are alive), and specify the Kafka topic partitions and initial offsets. > **NOTE** > > Because the actual task concurrency is determined by the minimum value of multiple parameters, you must make sure that the value of the FE dynamic parameter `max_routine_load_task_concurrent_num` is greater than or equal to `6`. ```sql ALTER ROUTINE LOAD FOR example_tbl2_ordertest2 PROPERTIES ( "desired_concurrent_number" = "6" ) FROM kafka ( "kafka_partitions" = "0,1,2,3,4,5,6,7", "kafka_offsets" = "OFFSET_BEGINNING,OFFSET_BEGINNING,OFFSET_BEGINNING,OFFSET_BEGINNING,OFFSET_END,OFFSET_END,OFFSET_END,OFFSET_END" ); ``` ## Stop a load job[​](#stop-a-load-job "Direct link to Stop a load job") You can execute the [STOP ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/STOP_ROUTINE_LOAD.md) statement to stop a load job. The state of the load job will be **STOPPED** after the statement is executed, and you cannot resume a stopped load job. You cannot check the status of a stopped load job with the [SHOW ROUTINE LOAD](https://docs.starrocks.io/docs/sql-reference/sql-statements/loading_unloading/routine_load/SHOW_ROUTINE_LOAD.md) statement. The following example stops the load job `example_tbl2_ordertest2`: ```sql STOP ROUTINE LOAD FOR example_tbl2_ordertest2; ```