E-MapReduce:Routine Load

Import CSV data

Step 1: Prepare your dataset

Suppose your Kafka topic my_topic contains CSV messages in this format:

val_k1,val_k2,val_k3,val_v1,val_v2,val_v3_raw

The six fields map to columns k1, k2, k3, v1, v2, and a raw value used to derive v3.

Step 2: Create a Routine Load job

CREATE ROUTINE LOAD example_db.test1 ON example_tbl
        COLUMNS TERMINATED BY ",",
        COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100)
        PROPERTIES
        (
            "desired_concurrent_number"="3",
            "max_batch_interval" = "20",
            "max_batch_rows" = "300000",
            "max_batch_size" = "209715200",
            "strict_mode" = "false"
        )
        FROM KAFKA
        (
            "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
            "kafka_topic" = "my_topic",
            "property.group.id" = "xxx",
            "property.client.id" = "xxx",
            "property.kafka_default_offsets" = "OFFSET_BEGINNING"
        );

Column mapping: Doris maps CSV fields to the COLUMNS list by position, then maps the listed column names to the table schema by name. The expression v3 = k1 * 100 is a transformation — Doris computes v3 from k1 rather than reading it directly from the source. If your source fields match the table columns exactly in name, number, and order, omit the COLUMNS clause.

OFFSET_BEGINNING starts consumption from the earliest available message. To start from the latest, use OFFSET_END.

Import CSV data in strict mode

Strict mode filters out rows where type conversion fails (for example, a string that cannot be cast to an integer). Use it when data quality is critical:

CREATE ROUTINE LOAD example_db.test1 ON example_tbl
        COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100),
        WHERE k1 > 100 and k2 like "%doris%"
        PROPERTIES
        (
            "desired_concurrent_number"="3",
            "max_batch_interval" = "20",
            "max_batch_rows" = "300000",
            "max_batch_size" = "209715200",
            "strict_mode" = "true"
        )
        FROM KAFKA
        (
            "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
            "kafka_topic" = "my_topic",
            "kafka_partitions" = "0,1,2,3",
            "kafka_offsets" = "101,0,0,200"
        );

kafka_partitions and kafka_offsets specify the starting offset for each partition. The number of entries in both parameters must match.

Strict mode behavior

The following tables show how strict mode affects rows with conversion failures.

*TinyInt column (nullable):*

*Decimal(1,0) column (nullable):*

The value 10 passes strict mode because it is a valid Decimal type value, but it is filtered during the extract, transform, load (ETL) phase because it exceeds the precision of Decimal(1,0).

Import JSON data

Routine Load supports two JSON layouts:

Single JSON object (one record per message):

{"category":"a9jadhx","author":"test","price":895}

JSON array (multiple records per message):

[
    {"category":"11","author":"4avc","price":895,"timestamp":1589191587},
    {"category":"22","author":"2avc","price":895,"timestamp":1589191487},
    {"category":"33","author":"3avc","price":342,"timestamp":1589191387}
]

Step 1: Create a target table

CREATE TABLE `example_tbl` (
   `category` varchar(24) NULL COMMENT "",
   `author` varchar(24) NULL COMMENT "",
   `timestamp` bigint(20) NULL COMMENT "",
   `dt` int(11) NULL COMMENT "",
   `price` double REPLACE
) ENGINE=OLAP
AGGREGATE KEY(`category`,`author`,`timestamp`,`dt`)
COMMENT "OLAP"
PARTITION BY RANGE(`dt`)
(
  PARTITION p0 VALUES [("-2147483648"), ("20200509")),
    PARTITION p20200509 VALUES [("20200509"), ("20200510")),
    PARTITION p20200510 VALUES [("20200510"), ("20200511")),
    PARTITION p20200511 VALUES [("20200511"), ("20200512"))
)
DISTRIBUTED BY HASH(`category`,`author`,`timestamp`) BUCKETS 4
PROPERTIES (
    "replication_num" = "1"
);

Step 2a: Import single-object JSON

CREATE ROUTINE LOAD example_db.test_json_label_1 ON table1
COLUMNS(category,price,author)
PROPERTIES
(
    "desired_concurrent_number"="3",
    "max_batch_interval" = "20",
    "max_batch_rows" = "300000",
    "max_batch_size" = "209715200",
    "strict_mode" = "false",
    "format" = "json"
)
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
    "kafka_topic" = "my_topic",
    "kafka_partitions" = "0,1,2",
    "kafka_offsets" = "0,0,0"
 );

Step 2b: Import JSON array

CREATE ROUTINE LOAD example_db.test1 ON example_tbl
COLUMNS(category, author, price, timestamp, dt=from_unixtime(timestamp, '%Y%m%d'))
PROPERTIES
(
    "desired_concurrent_number"="3",
    "max_batch_interval" = "20",
    "max_batch_rows" = "300000",
    "max_batch_size" = "209715200",
    "strict_mode" = "false",
    "format" = "json",
    "jsonpaths" = "[\"$.category\",\"$.author\",\"$.price\",\"$.timestamp\"]",
    "strip_outer_array" = "true"
)
FROM KAFKA
(
    "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
    "kafka_topic" = "my_topic",
    "kafka_partitions" = "0,1,2",
    "kafka_offsets" = "0,0,0"
);

Set strip_outer_array to true when messages contain a JSON array. Use jsonpaths to map specific JSON fields to table columns. The dt column is not in the source data, so the expression dt=from_unixtime(timestamp, '%Y%m%d') derives it from timestamp.

Connect to a Kafka cluster with SSL

If your Kafka cluster requires SSL authentication, upload the certificate files first, then reference them in the job.

Step 1: Upload certificate files

CREATE FILE "ca.pem" PROPERTIES("url" = "https://example_url/kafka-key/ca.pem", "catalog" = "kafka");
CREATE FILE "client.key" PROPERTIES("url" = "https://example_url/kafka-key/client.key", "catalog" = "kafka");
CREATE FILE "client.pem" PROPERTIES("url" = "https://example_url/kafka-key/client.pem", "catalog" = "kafka");

Upload the CA certificate (ca.pem) for all SSL connections. If the Kafka cluster also requires client authentication, upload client.pem and client.key as well. For help with the CREATE FILE command, run HELP CREATE FILE;.

Step 2: Create a Routine Load job with SSL

CREATE ROUTINE LOAD db1.job1 on tbl1
PROPERTIES
(
    "desired_concurrent_number"="1"
)
FROM KAFKA
(
    "kafka_broker_list"= "broker1:9091,broker2:9091",
    "kafka_topic" = "my_topic",
    "property.security.protocol" = "ssl",
    "property.ssl.ca.location" = "FILE:ca.pem",
    "property.ssl.certificate.location" = "FILE:client.pem",
    "property.ssl.key.location" = "FILE:client.key",
    "property.ssl.key.password" = "abcd***"
);

Doris connects to Kafka using the C++ client library librdkafka. For a full list of supported connection properties, see the librdkafka configuration reference.

View job status

SHOW ROUTINE LOAD;
SHOW ROUTINE LOAD TASK;

SHOW ROUTINE LOAD returns the job state and statistics. SHOW ROUTINE LOAD TASK shows the status of individual tasks within a job. Both commands display only jobs in the RUNNING state — completed and not-yet-started jobs are not shown.

For full syntax, run HELP SHOW ROUTINE LOAD; or HELP SHOW ROUTINE LOAD TASK;.

Modify a job

ALTER ROUTINE LOAD ...;

Use ALTER ROUTINE LOAD to update the properties of a job that has been created. For full syntax, run HELP ALTER ROUTINE LOAD;.

Pause, resume, or stop a job

For full syntax, run HELP PAUSE ROUTINE LOAD;, HELP RESUME ROUTINE LOAD;, or HELP STOP ROUTINE LOAD;.

When a paused job recovers automatically

Doris attempts automatic recovery for jobs paused by transient errors (such as a temporary BE failure). It retries up to three times within period_of_auto_resume_min (default: 5 minutes). If all three attempts fail, the job is locked and requires a manual RESUME.

Schema changes and imports

Routine Load jobs do not block SCHEMA CHANGE or ROLLUP operations. However, if a SCHEMA CHANGE causes column mismatches between the source data and the target table, the error row count increases and the job eventually pauses. To prevent this, specify explicit column mappings in the job and use NULLABLE columns or columns with a DEFAULT value.

Partition deletion

If you delete a table partition while a Routine Load job is importing into it, the job pauses because the partition can no longer be found.

Concurrent operations

Routine Load jobs run concurrently with LOAD and INSERT operations without conflict. Before running a DELETE on a table, pause the Routine Load job and wait for all assigned tasks to finish, then run DELETE.

Database and table deletion

If you drop the target database or table, the Routine Load job is canceled immediately.

Kafka topic auto-creation

If the Kafka topic specified in the job does not exist, Kafka brokers can create it automatically based on the auto.create.topics.enable setting:

• auto.create.topics.enable=true: Kafka creates the topic with the number of partitions defined by num.partitions. The job starts consuming normally.

• auto.create.topics.enable=false: Kafka does not create the topic. The job pauses until the topic exists and has data.

To ensure topics are auto-created, set auto.create.topics.enable=true on all brokers in the cluster.

Network isolation

In environments with CIDR-based isolation or restricted DNS:

• Doris must be able to reach all brokers in the kafka_broker_list.

• If the Kafka cluster has advertised.listeners configured, Doris must be able to reach those listener addresses, not just the broker list addresses.