Use Routine Load to import data

Strict mode behavior

Strict mode controls how type-conversion failures are handled.

Importing a TINYINT column (NULL allowed):

Importing a DECIMAL(1,0) column (NULL allowed):

The value 10 exceeds the DECIMAL(1,0) range but is not filtered by strict mode, because the value itself passes type conversion. It is filtered later during the extract, transform, and load (ETL) process.

Load CSV data

1. Create a destination table:

   CREATE TABLE test_table (
       id      INT,
       name    VARCHAR(50),
       age     INT,
       address VARCHAR(50),
       url     VARCHAR(500)
   )
   UNIQUE KEY(`id`, `name`)
   DISTRIBUTED BY HASH(id) BUCKETS 4
   PROPERTIES ("replication_num" = "1");

2. Create a Routine Load job that reads from the beginning of all partitions:

   CREATE ROUTINE LOAD example_db.test1 ON test_table
   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.kafka_default_offsets" = "OFFSET_BEGINNING"
   );

3. Create a Routine Load job with strict mode enabled:

   CREATE ROUTINE LOAD example_db.test2 ON test_table
   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"               = "true"
   )
   FROM KAFKA (
       "kafka_broker_list"              = "broker1:9092,broker2:9092,broker3:9092",
       "kafka_topic"                    = "my_topic",
       "property.kafka_default_offsets" = "OFFSET_BEGINNING"
   );

4. Start consuming from a specific timestamp:

   CREATE ROUTINE LOAD example_db.test4 ON test_table
   PROPERTIES (
       "desired_concurrent_number" = "3",
       "max_batch_interval"        = "30",
       "max_batch_rows"            = "300000",
       "max_batch_size"            = "209715200"
   )
   FROM KAFKA (
       "kafka_broker_list"             = "broker1:9092,broker2:9092",
       "kafka_topic"                   = "my_topic",
       "property.kafka_default_offset" = "2024-01-21 10:00:00"
   );

Load JSON data

Routine Load supports two JSON message structures:

• Single JSON object — one record per message:

  {"key1":"value1","key2":"value2","key3":"value3"}

• JSON array — multiple records per message:

  [
      {"key1":"value11","key2":"value12","key3":"value13","key4":14},
      {"key1":"value21","key2":"value22","key3":"value23","key4":24}
  ]

For multi-table import mode, prefix each message with the target table name:

table_name|{"key1":"value1","key2":"value2","key3":"value3"}

Example: load JSON data

1. Create a destination table:

   CREATE TABLE `example_tbl` (
       `category`  VARCHAR(24)  NULL,
       `author`    VARCHAR(24)  NULL,
       `timestamp` BIGINT(20)   NULL,
       `dt`        INT(11)      NULL,
       `price`     DOUBLE REPLACE
   ) ENGINE=OLAP
   AGGREGATE KEY(`category`, `author`, `timestamp`, `dt`)
   PARTITION BY RANGE(`dt`) (
       PARTITION p0        VALUES [("-2147483648"), ("20230509")),
       PARTITION p20200509 VALUES [("20230509"), ("20231010")),
       PARTITION p20200510 VALUES [("20231010"), ("20231211")),
       PARTITION p20200511 VALUES [("20231211"), ("20240512"))
   )
   DISTRIBUTED BY HASH(`category`, `author`, `timestamp`) BUCKETS 4;

2. Publish both message types to a Kafka topic:

   {"category":"value1331","author":"value1233","timestamp":1700346050,"price":1413}

   [
       {"category":"value13z2","author":"vaelue13","timestamp":1705645251,"price":14330},
       {"category":"lvalue211","author":"lvalue122","timestamp":1684448450,"price":24440}
   ]

3. Load in simple mode (field names match column names):

   CREATE ROUTINE LOAD example_db.test_json_label_1 ON example_tbl
   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"
   );

4. Load with explicit JSON path extraction and outer array stripping:

   The dt column does not appear in the source data. Its value is derived from timestamp using dt=from_unixtime(timestamp,'%Y%m%d') in the COLUMNS clause.

   CREATE ROUTINE LOAD example_db.test_json_label_3 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"
   );

Connect to Kafka clusters with authentication

ApsaraDB for SelectDB uses the librdkafka C++ client library to connect to Kafka. For supported configuration properties, see the librdkafka configuration reference.

SSL authentication

Upload the required certificate files first, then create the Routine Load job.

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");

2. Create the Routine Load job:

   Property	Required	Description
   property.security.protocol	Always	Set to ssl.
   property.ssl.ca.location	Always	Path to the CA certificate that authenticates Kafka broker public keys.
   property.ssl.certificate.location	Only when client authentication is enabled on the Kafka server	Path to the client public key certificate.
   property.ssl.key.location	Only when client authentication is enabled on the Kafka server	Path to the client private key file.
   property.ssl.key.password	Only when client authentication is enabled on the Kafka server	Password for the client private key.

   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"         = "abcdefg"
   );

PLAIN authentication

CREATE ROUTINE LOAD db1.job1 ON tbl1
PROPERTIES (
    "desired_concurrent_number" = "1"
)
FROM KAFKA (
    "kafka_broker_list"          = "broker1:9092,broker2:9092",
    "kafka_topic"                = "my_topic",
    "property.security.protocol" = "SASL_PLAINTEXT",
    "property.sasl.mechanism"    = "PLAIN",
    "property.sasl.username"     = "admin",
    "property.sasl.password"     = "admin"
);

Kerberos authentication

Before creating the job, deploy the Kerberos client kinit on all nodes in your ApsaraDB for SelectDB cluster, configure krb5.conf, and specify the Key Distribution Center (KDC) service.

CREATE ROUTINE LOAD db1.job1 ON tbl1
PROPERTIES (
    "desired_concurrent_number" = "1"
)
FROM KAFKA (
    "kafka_broker_list"                   = "broker1:9092,broker2:9092",
    "kafka_topic"                         = "my_topic",
    "property.security.protocol"          = "SASL_PLAINTEXT",
    "property.sasl.kerberos.service.name" = "kafka",
    "property.sasl.kerberos.keytab"       = "/etc/krb5.keytab",
    "property.sasl.kerberos.principal"    = "id@your.com"
);