E-MapReduce:Data import FAQ

"close index channel failed" or "too many tablet versions" error

• Cause

  A high data import frequency prevents timely compaction, causing the number of un-compacted tablet versions to exceed the maximum supported limit.

• Solution

  By default, the maximum number of supported un-compacted versions is 1,000. To resolve this issue:

  • Increase the data volume of each import to reduce the data import frequency.

  • Modify the following parameters in each backend's (BE) be.conf configuration file to adjust the compaction policy and accelerate compaction.

    cumulative_compaction_num_threads_per_disk = 4
    base_compaction_num_threads_per_disk = 2
    cumulative_compaction_check_interval_seconds = 2

Handling the "Label Already Exists" error

• Problem description

  An import job with the same label has already completed or is currently running within the same database in a StarRocks cluster.

• Cause

  Stream Load uses the HTTP protocol to submit import job requests, and most HTTP clients have a built-in retry mechanism. After a StarRocks cluster receives the first request, it starts processing the Stream Load job. If the cluster does not respond to the client before the client's timeout period ends, the client may automatically resend the same request. Because the StarRocks cluster is already processing the original request, it rejects the second request and returns the Label Already Exists error.

• Solution

  Check for label conflicts between different import methods or for duplicate import job submissions. You can diagnose the issue as follows:

  • Search the log of the primary frontend (FE) node for the job's label. If the label appears twice, the client has sent a duplicate request.

    Note

    In a StarRocks cluster, import job labels are not scoped by import method. Therefore, different types of import jobs can conflict if they use the same label.

  • Run the SHOW LOAD WHERE LABEL = "xxx" statement to check if an import job with the same label is in the FINISHED state. Replace xxx with your label.

  To prevent this issue, estimate the import duration based on the data volume of your request. Then, set a long enough request timeout period on the client to avoid premature retries.

Resolving the "ETL_QUALITY_UNSATISFIED; msg:quality not good enough to cancel" error

Error: Failed to find enough host

You can add "replication_num" = "1" in the table properties.

Resolve the "Too many open files" error

The "increase config load_process_max_memory_limit_percent" error

/root/starrocks/be/src/runtime/plan_fragment_executor.cpp:209 _sink->open(runtime_state())
W0516 15:58:30.892597 28779 fragment_mgr.cpp:193] Fail to open fragment 334f7f8c-0727-5909-afa3-9eb1dxxx : Internal error: intolerable failure in opening node channels
/root/starrocks/be/src/runtime/plan_fragment_executor.cpp:209 _sink->open(runtime_state())
W0516 15:58:30.898306 28779 stream_load_executor.cpp:94] fragment execute failed, query_id=334f7f8c07275909-afa39eb1dxxx, err_msg=intolerable failure in opening node channels, id=334f7f8c07275909-afa3
9eb1db88af94, job_id=-1, txn_id=633657, label=156936fa-1134-4c4c-b473-fa05b4b8bf8c
W0516 15:58:30.907560 28985 stream_load.cpp:315] append body content failed. errmsg=cancelledid=334f7f8c07275909-afa39eb1dbxxx, job_id=-1, txn_id=633657, label=156936fa-1134-4c4c-b473-fa05b4b8bf8c
W0516 15:58:34.353019 28993 data_sink.cpp:104] Ignore option use_vectorized=false
W0516 15:58:34.353847 28795 tablet_sink.cpp:605] NodeChannel[15381-10004]: tablet open failed, load_id=cb4e78a6-f026-c3d9-a700-aba4xxx, txn_id=6336xxx, node=10.28.xxx, 8060, errmsg=memory limit exc
eeded, please reduce load frequency or increase config `load_process_max_memory_limit_percent` or add more BE nodes
W0516 15:58:34.353901 28795 tablet_sink.cpp:605] NodeChannel[15381-10003]: tablet open failed, load_id=cb4e78a6-f026-c3d9-a700-aba4xxx, txn_id=6336xxx, node=10.28.xxx, 8060, errmsg=memory limit exc
eeded, please reduce load frequency or increase config `load_process_max_memory_limit_percent` or add more BE nodes
W0516 15:58:34.353909 28795 tablet_sink.cpp:605] NodeChannel[15381-149009]: tablet open failed, load_id=cb4e78a6-f026-c3d9-a700-abaxxx, txn_id=633xxx, node=10.2xxx, 8060, errmsg=memory limit exc
eeded, please reduce load frequency or increase config `load_process_max_memory_limit_percent` or add more BE nodes
W0516 15:58:34.353914 28795 tablet_sink.cpp:605] NodeChannel[15381-10002]: tablet open failed, load_id=cb4e78a6-f026-c3d9-a700-aba4xxx, txn_id=6336xxx, node=10.28.xxx, 8060, errmsg=memory limit exc
eeded, please reduce load frequency or increase config `load_process_max_memory_limit_percent` or add more BE nodes
W0516 15:58:34.353976 28795 tablet_sink.cpp:605] NodeChannel[15381-149010]: tablet open failed, load_id=cb4e78a6-f026-c3d9-a700-abaxxx, txn_id=633xxx, node=10.2xxx, 8060, errmsg=memory limit exc
eeded, please reduce load frequency or increase config `load_process_max_memory_limit_percent` or add more BE nodes
W0516 15:58:34.353983 28795 tablet_sink.cpp:613] open failed, load_id=cb4e78a6f026c3d9-a700aba44764c288
W0516 15:58:34.353992 28795 plan_fragment_executor.cpp:188] fail to open fragment, instance_id=cb4e78a6-f026-c3d9-a700-aba44xxx, status=Internal error: intolerable failure in opening node channels
/root/starrocks/be/src/runtime/plan_fragment_executor.cpp:209 _sink->open(runtime_state())
W0516 15:58:34.354202 28795 fragment_mgr.cpp:193] Fail to open fragment cb4e78a6-f026-c3d9-a700-aba447xxx : Internal error: intolerable failure in opening node channels
/root/starrocks/be/src/runtime/plan_fragment_executor.cpp:209 _sink->open(runtime_state())
W0516 15:58:34.600633 28795 stream_load_executor.cpp:94] fragment execute failed, query_id=cb4e78a6f026c3d9-a700aba44xxx, err_msg=intolerable failure in opening node channels, id=cb4e78a6f026c3d9-a700
aba44764c288, job_id=-1, txn_id=633662, label=156936fa-1134-4c4c-b473-fa05b4b8bf8c
W0516 15:58:34.600664 28993 stream_load.cpp:132] Fail to handle streaming load, id=cb4e78a6f026c3d9-a700aba447xxx errmsg=intolerable failure in opening node channels
W0516 15:58:39.555845 28995 data_sink.cpp:104] Ignore option use_vectorized=false

RPC timeouts during data import

Check the write_buffer_size parameter in the BE's be.conf file. This parameter sets the size threshold for a memory block on the BE and defaults to 100 MB. If this threshold is too large, it may cause an RPC timeout. To resolve this, adjust the write_buffer_size and tablet_writer_rpc_timeout_sec parameters. For more information about BE configuration parameters, see Parameter settings.

Error: "Value count does not match column count"

• Problem description

  An import job fails, and the error details URL shows an error: "Value count does not match column count". This indicates that the number of columns in the source data does not match the number of columns in the destination table.

  Error: Value count does not match column count. Expect 3, but got 1. Row: 2023-01-01T18:29:00Z,cpu0,80.99
  Error: Value count does not match column count. Expect 3, but got 1. Row: 2023-01-01T18:29:10Z,cpu1,75.23
  Error: Value count does not match column count. Expect 3, but got 1. Row: 2023-01-01T18:29:20Z,cpu2,59.44

• Cause

  This error occurs when the column delimiter specified in the import command or import statement does not match the delimiter in the source data. In this example, the source data has three columns and uses a comma (,) as its column delimiter. However, if the import command or statement specifies a tab character (\t) as the delimiter, the three columns of source data are incorrectly parsed as a single column.

• Solution

  Change the column delimiter in your import command or import statement to a comma (,), and retry the import job.

Choosing an import method

For more information, see Data Import.

Factors affecting import performance

Skip header rows in Stream Load

Non-standard partition key in Stream Load

StarRocks supports data conversion during the import process.

-H "columns: NO,DATE_1, VERSION, PRICE, DATE=LEFT(DATE_1,6)"

In this example, DATE_1 acts as a temporary placeholder for the source data. The LEFT() function then transforms this data, and the result is assigned to the DATE column in the StarRocks table. Note that you must map all columns from the source CSV file to temporary names before applying a transformation function. Any function used for column transformation must be a scalar function, such as a non-aggregate function or a window function.

"body exceed max size: 10737418240, limit: 10737418240" error

• Cause

  The source data file exceeds the maximum file size (10 GB) supported by Stream Load.

• Solution

  • Split the source data file into smaller files using the seq -w 0 n command.

  • Run the curl -XPOST http:///be_host:http_port/api/update_config?streaming_load_max_mb=<file_size> command to increase the file size limit by modifying the streaming_load_max_mb parameter on the BE node. For details on BE parameter settings, see parameter settings.

Improve import performance

Method 1: Increase task parallelism

min(alive_be_number, partition_number, desired_concurrent_number, max_routine_load_task_concurrent_num)

Therefore, if you need to increase the actual task parallelism when the number of partitions and BE nodes is large and exceeds the desired_concurrent_number and max_routine_load_task_concurrent_num parameters, you can increase the desired_concurrent_number and max_routine_load_task_concurrent_num parameters.

For example, assume you have 7 partitions and 5 active BE nodes, and max_routine_load_task_concurrent_num is set to its default value of 5. To maximize the actual task parallelism, you must set desired_concurrent_number to 5 (the default is 3). The actual task parallelism is then calculated as min(5,7,5,5), which equals 5.

Method 2: Increase data volume per import task

Two FE dynamic parameters, max_routine_load_batch_size and routine_load_task_consume_second, control the maximum data volume an import task can consume. An import task finishes when the threshold for either parameter is reached. For configuration details, see parameter settings.

I0325 20:27:50.410579 15259 data_consumer_group.cpp:131] consumer group done: 41448fb1a0ca59ad-30e34dabfa7e47a0. consume time(ms)=3261, received rows=179190, received bytes=9855450, eos: 1, left_time: -261, left_bytes: 514432550, blocking get time(us): 3065086, blocking put time(us): 24855

Normally, the log shows left_bytes >= 0. This indicates that an import batch did not reach the max_routine_load_batch_size limit before the time limit from routine_load_task_consume_second expired. This means the import tasks are keeping up with the data stream from Kafka, resulting in no consumption latency. In this case, you can increase routine_load_task_consume_second to increase the data volume consumed by each import task.

If left_bytes < 0, it indicates that a batch reached the max_routine_load_batch_size limit before its time limit (routine_load_task_consume_second) expired. This suggests that each import batch is filling to capacity, which can lead to a data backlog from Kafka. In this scenario, you can increase max_routine_load_batch_size.

Troubleshooting PAUSED and CANCELLED import jobs

Routine Load and exactly-once semantics

Yes, Routine Load guarantees exactly-once semantics.

Each import task executes as a single transaction. If an import task fails, the transaction is aborted, and the FE does not update the consumption progress for the related partitions. When the FE schedules the next import task from the task execution queue, it initiates a consumption request from the last saved consumer offset of the partition. This process guarantees exactly-once semantics.

"Broker: Offset out of range" error

Rerunning Broker Load jobs in FINISHED state

No. You cannot rerun an import job in the FINISHED state. To ensure exactly-once semantics, you cannot reuse the label of a successful import job. You can use the SHOW LOAD statement to view your import history, find the job you want to rerun, copy its configuration, assign a new label, and create and run a new import job.

Date field 8-hour offset with Broker Load

• Cause

  This occurs because of a time zone mismatch. The StarRocks table and the Broker Load import job use the China time zone (UTC+8), while the server uses UTC. This difference causes the import process to add an 8-hour offset to the date field.

• Solution

  Remove the timezone parameter when you create the table.

"ErrorMsg: type:ETL_RUN_FAIL; msg:Cannot cast '<slot 6>' from VARCHAR to ARRAY<VARCHAR(30)>" error during Broker Load of ORC data

• Cause

  This error is caused by a mismatch between the column names in the source data file and the destination StarRocks table. This mismatch triggers a SET statement, leading to a data type conversion failure within the cast function.

• Solution

  Ensure that the column names in the source file match those in the StarRocks table. This eliminates the need for a SET statement and data type conversion, allowing the import to succeed.

No data found after Broker Load

Broker Load is an asynchronous import method. Successfully submitting an import job does not mean the import was successful. You can run the SHOW LOAD statement to check the final status of the import job and the error message in the errmsg field. If the job fails, modify its parameters and retry it.

Error: "failed to send batch" or "TabletWriter add batch with unknown id"

This error indicates a data write timeout. To resolve this, modify the system variable query_timeout and the BE parameter streaming_load_rpc_max_alive_time_sec. For more information about BE parameters, see Parameter Settings.

"LOAD-RUN-FAIL; msg:OrcScannerAdapter::init_include_columns. col name = xxx not found" error

When importing data from a Parquet or ORC file, ensure the column names in the file header match those in the StarRocks table.

(tmp_c1,tmp_c2)
SET
(
   id=tmp_c2,
   name=tmp_c1
)

Importing an ORC file generated by Apache Hive can cause an "Invalid Column Name" error if its table header is (_col0, _col1, _col2, ...). In this case, you must use the SET clause to define the column mapping rules.

Troubleshoot long-running import jobs

In the fe.log file on the FE, find the import job ID using its label. Then, in the be.INFO file on the BE, use this ID to find the relevant log entries and identify the cause.

Configure ViewFs for HDFS Federation

Copy the configuration files, core-site.xml and hdfs-site.xml, to the broker/conf directory.

If you use a custom file system, also copy the corresponding .jar files to the broker/lib directory.

"Can't get Kerberos realm" error

1. Verify that the /etc/krb5.conf file is configured on all nodes where the brokers are running.

2. If the error persists, append -Djava.security.krb5.conf:/etc/krb5.conf to the JAVA_OPTS variable in the broker's startup script.

Slow performance of single-record inserts

The INSERT INTO statement is designed for batch writes. A single write takes as long as a batch write. Therefore, in an OLAP scenario, do not use an INSERT INTO statement for single writes.

"index channel has intolerable failure" error

This error occurs due to a Stream Load RPC timeout. To resolve this, adjust the RPC timeout parameters in the configuration file.

INSERT INTO SELECT fails with execute timeout

This error is caused by a query timeout. You can resolve this issue by adjusting the session variable query_timeout. The default value for this parameter is 600 seconds.

set query_timeout =xx;

Flink job error: Required options are missing

• Cause

  This error occurs because multiple rules, such as [table-rule.1] and [table-rule.2], in the config_prod.conf configuration file of StarRocks-migrate-tools (SMT) lack required configuration information.

• Solution

  Verify that database, table, and Flink connector information is configured for each rule, such as [table-rule.1] and [table-rule.2].

Automatic restart of failed tasks

Flink uses a checkpointing mechanism and a restart policy to automatically restart failed tasks.

# unit: ms
execution.checkpointing.interval: 300000
state.backend: filesystem
state.checkpoints.dir: file:///tmp/flink-checkpoints-directory

Stop and restore a Flink job

When stopping a Flink job, you can manually trigger a savepoint. A savepoint is a consistent snapshot of a streaming job's execution state, created by the checkpointing mechanism. You can later restore the job from a savepoint.

To stop a job with a savepoint, Flink first triggers a savepoint for the job, and then stops it. You can also specify a target directory to store the savepoint.

bin/flink stop --type [native/canonical] --savepointPath [:targetDirectory] :jobId

./flink run -c com.starrocks.connector.flink.tools.ExecuteSQL -s savepoints_dir/savepoints-xxxxxxxx flink-connector-starrocks-xxxx.jar -f flink-create.all.sql

Data import failure with exactly-once

• Description: The system reports the following error.

  com.starrocks.data.load.stream.exception.StreamLoadFailException: {
      "TxnId": 3382****,
      "Label": "502c2770-cd48-423d-b6b7-9d8f9a59****",
      "Status": "Fail",
      "Message": "timeout by txn manager",-- Error message
      "NumberTotalRows": 1637,
      "NumberLoadedRows": 1637,
      "NumberFilteredRows": 0,
      "NumberUnselectedRows": 0,
      "LoadBytes": 4284214,
      "LoadTimeMs": 120294,
      "BeginTxnTimeMs": 0,
      "StreamLoadPlanTimeMs": 7,
      "ReadDataTimeMs": 9,
      "WriteDataTimeMs": 120278,
      "CommitAndPublishTimeMs": 0
  }

• Cause: The value of sink.properties.timeout is less than the Flink checkpoint interval, causing a transaction timeout.

• Solution: Set this parameter to a value greater than the Flink checkpoint interval.

Time lag for flink-connector-jdbc_2.11 sink to StarRocks

• Problem description: The time generated by the localtimestamp function in Flink is correct, but it is 8 hours behind after being written to StarRocks. Both the Flink and StarRocks servers are set to the Asia/Shanghai time zone (UTC+8). The Flink version is 1.12 and the driver is flink-connector-jdbc_2.11.

• Solution: To resolve this issue, configure the time zone in the Flink sink table. Set the parameter 'server-time-zone' = 'Asia/Shanghai' and append &serverTimezone=Asia/Shanghai to the url parameter. The following example shows this configuration.

  CREATE TABLE sk (
      sid int,
      local_dtm TIMESTAMP,
      curr_dtm TIMESTAMP
  )
  WITH (
      'connector' = 'jdbc',
      'url' = 'jdbc:mysql://192.168.**.**:9030/sys_device?characterEncoding=utf-8&serverTimezone=Asia/Shanghai',
      'table-name' = 'sink',
      'driver' = 'com.mysql.jdbc.Driver',
      'username' = 'sr',
      'password' = 'sr123',
      'server-time-zone' = 'Asia/Shanghai'
  );

Cannot import from an external Kafka cluster

• Description: The import fails with the following error message:

  failed to query wartermark offset, err: Local: Bad message format

• Cause: The StarRocks nodes cannot resolve the hostnames of the Kafka brokers.

• Solution: Add the hostnames and IP addresses of the Kafka brokers to the /etc/hosts file on each StarRocks node.

High BE memory and CPU usage when idle

The BE node periodically collects statistics, causing temporary spikes in CPU usage rather than sustained high utilization. Additionally, the BE node retains and internally manages up to 10 GiB of used memory instead of releasing it to the operating system. You can configure the threshold for this memory retention with the tc_use_memory_min parameter.

tc_use_memory_min specifies the minimum memory that TCmalloc reserves. The default value is 10,737,418,240 (10 GiB). StarRocks returns idle memory to the operating system only when the used memory exceeds this value. You can configure this parameter in the be.conf file on the Configure tab of the StarRocks service in the EMR console. For more information about BE node parameters, refer to parameter settings.

Why BE does not release memory

This is expected behavior. A database allocates large blocks of memory from the operating system—often pre-allocating more than immediately needed—and retains this memory for reuse. This strategy avoids the performance overhead of frequent, costly memory re-allocations. To verify this behavior, monitor memory usage in a test environment over an extended period to see if the memory is eventually released.

Cannot resolve Flink connector dependencies

• Cause: Flink connector dependencies must be fetched from an Alibaba Cloud mirror. The error occurs because the mirror section in the /etc/maven/settings.xml file is not configured to direct all repository requests to this mirror.

• Solution: In /etc/maven/settings.xml, configure the mirror section to use the Alibaba Cloud public repository: https://maven.aliyun.com/repository/public.

Effectiveness of the sink.buffer-flush.interval-ms parameter

• Problem description: When sink.buffer-flush.interval-ms is set to 15s and checkpoint interval=5mins, does sink.buffer-flush.interval-ms still take effect?

  +----------------------+--------------------------------------------------------------+
  |         Option       | Required |  Default   | Type   |       Description           |
  +-------------------------------------------------------------------------------------+
  |  sink.buffer-flush.  |  NO      |   300000   | String | the flushing time interval, |
  |  interval-ms         |          |            |        | range: [1000ms, 3600000ms]  |
  +----------------------+--------------------------------------------------------------+

• Solution: A flush is triggered when the first of the following three thresholds is met. This behavior is independent of the checkpoint interval. The checkpoint interval is only for exactly-once semantics, while sink.buffer-flush.interval-ms is for at-least-once semantics.

  sink.buffer-flush.max-rows
  sink.buffer-flush.max-bytes
  sink.buffer-flush.interval-ms

Data updates with DataX

DataX now supports data updates for the primary key model. To enable this, add the _op field to the reader section in your JSON configuration file.

Handling keywords in field names

Enclose the field name in backticks ().

HADOOP-CONF-DIR or YARN-CONF-DIR must be set

This error indicates that the HADOOP-CONF-DIR environment variable, required for Spark Load, has not been set in the spark-env.sh file on the spark client.

spark-submit command fails with "No such file or directory" error

This error occurs when using Spark Load because the spark_home_default_dir parameter is either unspecified or set to an incorrect Spark client root directory.

"File xxx/jars/spark-2x.zip does not exist" error

When you use Spark Load, this error indicates that the spark-resource-path parameter does not point to the packaged ZIP file. Verify the file path and file name.

"Yarn client does not exist" error

This error indicates that the yarn-client-path parameter does not specify the path to the YARN executable file.