StarRocks Routine Load: Import Kafka JSON & CSV Data with Real-World Troubleshooting

Environment Preparation

Basic requirements:

Kafka cluster version 0.10.2 or higher, with test topics created.

StarRocks cluster version 2.0 or higher, including FE and BE (CN) nodes.

JDK 1.8+ with environment variables configured.

Network connectivity: StarRocks must reach Kafka broker ports (default 9092).

Topic and Data Preparation

Create two Kafka topics for JSON and CSV data:

# Create topic for JSON data
kafka-topics.sh --create --bootstrap-server kafka-broker1:9092,kafka-broker2:9092 --topic user_behavior_json --partitions 3 --replication-factor 2
# Create topic for CSV data
kafka-topics.sh --create --bootstrap-server kafka-broker1:9092,kafka-broker2:9092 --topic order_info_csv --partitions 2 --replication-factor 2

JSON Format Data Import Example

Sample Data

Kafka topic user_behavior_json stores user behavior events as JSON lines, each containing user_id, behavior_type, product_id, timestamp, and a nested details object (page, stay_time, device).

Create Target Table in StarRocks

Table user_behavior uses Primary Key model for frequent updates. The nested details field maps to a STRUCT type:

CREATE TABLE user_behavior (
  user_id BIGINT COMMENT '用户ID',
  behavior_type STRING COMMENT '行为类型，如click、purchase等',
  product_id STRING COMMENT '商品ID',
  timestamp BIGINT COMMENT '时间戳',
  details STRUCT<
    page STRING COMMENT '页面名称',
    stay_time INT COMMENT '停留时间(秒)',
    device STRING COMMENT '设备类型'
  > COMMENT '详细信息'
) ENGINE=OLAP
PRIMARY KEY(user_id, timestamp)
DISTRIBUTED BY HASH(user_id) BUCKETS 8
PROPERTIES (
  "replication_num" = "1",
  "storage_medium" = "SSD"
);

Create Routine Load Job for JSON

Key configuration: jsonpaths maps JSON fields to table columns in order; read_json_by_line=true parses one JSON object per line; desired_concurrent_number=3 matches Kafka partitions.

CREATE ROUTINE LOAD db_name.json_load_task ON user_behavior
COLUMNS(
  user_id,
  behavior_type,
  product_id,
  timestamp,
  details
),
COLUMNS FROM PATH AS (json_data)
PROPERTIES (
  "format" = "json",
  "jsonpaths" = "[
    \"$.user_id\",
    \"$.behavior_type\",
    \"$.product_id\",
    \"$.timestamp\",
    \"$.details\"
  ]",
  "read_json_by_line" = "true",
  "desired_concurrent_number" = "3",
  "max_batch_interval" = "20",
  "max_error_number" = "1000",
  "strict_mode" = "false"
)
FROM KAFKA (
  "kafka_broker_list" = "kafka-broker1:9092,kafka-broker2:9092",
  "kafka_topic" = "user_behavior_json",
  "kafka_consumer_group" = "sr_json_consumer_group",
  "kafka_partitions" = "0,1,2",
  "kafka_offsets" = "OFFSET_BEGINNING"
);

Parameter Explanation

format : "json"

jsonpaths : Defines mapping from JSON fields to table columns; missing this caused the initial null-field error.

read_json_by_line : true = each line is a separate JSON object.

desired_concurrent_number : Concurrency set to 3, matching the topic's 3 partitions.

max_batch_interval : Max seconds per batch (20s).

kafka_offsets : OFFSET_BEGINNING starts from earliest message.

Verify JSON Import

SELECT user_id, behavior_type, product_id, timestamp, details.page, details.stay_time
FROM user_behavior
LIMIT 10;

Successful query shows imported user behavior data.

CSV Format Data Import Example

Sample Data

Topic order_info_csv contains comma-separated lines: order_id, user_id, amount, order_time, status.

Create Target Table

CREATE TABLE order_info (
  order_id STRING COMMENT '订单ID',
  user_id BIGINT COMMENT '用户ID',
  amount DECIMAL(10, 2) COMMENT '订单金额',
  order_time DATETIME COMMENT '订单时间',
  status STRING COMMENT '订单状态'
) ENGINE=OLAP
PRIMARY KEY(order_id, user_id)
DISTRIBUTED BY HASH(order_id) BUCKETS 6
PROPERTIES (
  "replication_num" = "1"
);

Create Routine Load Job for CSV

CREATE ROUTINE LOAD db_name.csv_load_task ON order_info
COLUMNS(
  order_id,
  user_id,
  amount,
  order_time,
  status
)
PROPERTIES (
  "format" = "csv",
  "column_separator" = ",",
  "line_delimiter" = "
",
  "desired_concurrent_number" = "2",
  "max_batch_interval" = "15",
  "max_error_number" = "500",
  "skip_header" = "false"
)
FROM KAFKA (
  "kafka_broker_list" = "kafka-broker1:9092,kafka-broker2:9092",
  "kafka_topic" = "order_info_csv",
  "kafka_consumer_group" = "sr_csv_consumer_group",
  "kafka_partitions" = "0,1",
  "kafka_offsets" = "OFFSET_BEGINNING"
);

CSV Parameter Explanation

format : "csv"

column_separator : Comma.

line_delimiter : Newline.

desired_concurrent_number : 2, matching the 2 partitions.

skip_header : false (no header row to skip).

Verify CSV Import

SELECT * FROM order_info LIMIT 10;

Routine Load Task Management and Monitoring

View Task Status

SHOW ROUTINE LOAD \G;

Displays task name, state, progress, error details.

Pause and Resume Tasks

PAUSE ROUTINE LOAD FOR json_load_task;
PAUSE ROUTINE LOAD FOR csv_load_task;
RESUME ROUTINE LOAD FOR json_load_task;
RESUME ROUTINE LOAD FOR csv_load_task;

Stop Task (Irreversible)

STOP ROUTINE LOAD FOR json_load_task;
STOP ROUTINE LOAD FOR csv_load_task;

View Error Data

show routine load;
show routine load task where JobName='csv_load_task';

Notes and Optimization Recommendations

Data Format Validation

JSON: Validate with tools like jsonlint to avoid syntax errors.

CSV: Ensure consistent field separators; wrap fields containing the separator in quotes.

Performance Tuning

Set desired_concurrent_number equal to or a multiple of Kafka partition count.

Adjust max_batch_size and max_batch_interval to balance latency vs. throughput: lower interval for real-time, larger batch size for high throughput.

Allocate sufficient memory and CPU to BE nodes to avoid bottlenecks.

Fault Tolerance

Set reasonable max_error_number to tolerate some bad records without stopping the job.

Monitor task status regularly and handle failures promptly.

Enable replication via replication_num for critical data durability.

Summary

Based on a real troubleshooting case (missing jsonpaths causing nulls in a primary key table), this article walks through end-to-end ingestion of JSON and CSV from Kafka into StarRocks using Routine Load. It covers environment setup, table schema design (including STRUCT for nested JSON), load job creation with all key parameters, verification queries, task lifecycle commands, error inspection, and practical tuning advice. The combination of Kafka and StarRocks enables a powerful real-time analytics platform.

Additional: Flattening JSON to a 2D Table

To map nested JSON fields directly to flat columns, define each nested field as a separate column in the table and extend jsonpaths accordingly:

CREATE TABLE user_behavior (
  user_id BIGINT COMMENT '用户ID',
  behavior_type STRING COMMENT '行为类型，如click、purchase等',
  product_id STRING COMMENT '商品ID',
  timestamp BIGINT COMMENT '时间戳',
  details_page STRING COMMENT '页面名称',
  details_stay_time INT COMMENT '停留时间(秒)',
  details_device STRING COMMENT '设备类型'
) ENGINE=OLAP
PRIMARY KEY(user_id, timestamp)
DISTRIBUTED BY HASH(user_id) BUCKETS 8
PROPERTIES (
  "replication_num" = "1",
  "storage_medium" = "SSD"
);

CREATE ROUTINE LOAD db_name.json_load_task ON user_behavior
COLUMNS(
  user_id,
  behavior_type,
  product_id,
  timestamp,
  details_page,
  details_stay_time,
  details_device
),
COLUMNS FROM PATH AS (json_data)
PROPERTIES (
  "format" = "json",
  "jsonpaths" = "[
    \"$.user_id\",
    \"$.behavior_type\",
    \"$.product_id\",
    \"$.timestamp\",
    \"$.details.page\",
    \"$.details.stay_time\",
    \"$.details.device\"
  ]",
  "read_json_by_line" = "true",
  "desired_concurrent_number" = "3",
  "max_batch_interval" = "20",
  "max_error_number" = "1000",
  "strict_mode" = "false"
)
FROM KAFKA (
  "kafka_broker_list" = "kafka-broker1:9092,kafka-broker2:9092",
  "kafka_topic" = "user_behavior_json",
  "kafka_consumer_group" = "sr_json_consumer_group",
  "kafka_partitions" = "0,1,2",
  "kafka_offsets" = "OFFSET_BEGINNING"
);