CREATE ROUTINE LOAD

Purpose

This statement is used to create a routine load task that continuously consumes data from Kafka and imports it into tables in OceanBase Database.

Privilege requirements

To create a routine load job, the current user must have the INSERT privilege on the target table. For more information about OceanBase Database privileges, see Privilege types in MySQL-compatible mode.

Syntax

CREATE ROUTINE LOAD job_name ON table_name
    [load_property_list]
    FROM STREAM (
        [TYPE = 'KAFKA']
        STRING = STRING
        FORMAT [=] (format_options));

load_property_list:
    PARALLEL [=] int_value
    | COLUMNS (col1, col2, ...)
    | PARTITION (partition_name_list)
    | {IGNORE | REPLACE}
    | WHERE condition

STRING = STRING:
    'bootstrap.servers' = 'string_value',
    'kafka_topic' = 'string_value',
    ['kafka_partitions' = 'string_value',]
    ['kafka_offsets' = 'string_value',]
    ['max_batch_interval' = 'string_value',]
    ['max_batch_rows' = 'string_value',]
    ['max_batch_size' = 'string_value',]

format_options:
    TYPE = 'CSV',
    LINE_DELIMITER = '<string>' | <expr>,
    FIELD_DELIMITER = '<string>' | <expr>,
    ESCAPE = '<character>' | <expr>,
    FIELD_OPTIONALLY_ENCLOSED_BY = '<character>' | <expr>,
    ENCODING = 'charset',
    NULL_IF = ('<string>' | <expr>, '<string>' | <expr> ...),
    SKIP_BLANK_LINES = {TRUE | FALSE},
    TRIM_SPACE = {TRUE | FALSE},
    EMPTY_FIELD_AS_NULL = {TRUE | FALSE}

Parameters

load_property_list

• PARALLEL [=] int_value: Specifies the degree of parallelism. Default value: 1.

• COLUMNS (col1, col2, ...) : The mapping between Kafka message columns and target table columns.

• PARTITION (partition_name_list): Specifies the target partition for import.

• {IGNORE | REPLACE}: Specifies how to handle conflicts with primary keys or unique keys.

  • IGNORE: ignores the conflicting row.

• WHERE condition: Specifies the data filtering condition, which filters the data before writing it to the target table.

kafka_expr_list

• bootstrap.servers: Required. Specifies the list of Kafka broker addresses, separated by commas. For example, broker1:9092,broker2:9092. This is the entry address for the client to connect to Kafka for the first time.
• kafka_topic: Required. The name of the Kafka topic to consume.
• kafka_partitions: Optional. Specifies the list of partitions to consume, for example, "0,1,2". If not provided, all partitions under the topic will be consumed.
• kafka_offsets: Optional. Specifies the starting consumption offset for each partition, which corresponds one-to-one with kafka_partitions. That is, it specifies the initial offset for each partition. If only one value is provided, it applies to all partitions. If not specified, the subscription will start from OFFSET_END for all partitions under the topic by default. Currently, numbers and the special values OFFSET_BEGINNING and OFFSET_END are supported.
• "max_batch_interval" : Optional. The maximum running time per import thread in each round of a job, in seconds. The default value is 60 seconds (the number of import threads depends on the concurrency level PARALLEL).
• "max_batch_rows" : Optional. The maximum number of rows that each import thread reads in each batch. This value must be greater than or equal to 200,000. Default: 10,000,000.
• "max_batch_size": Optional. The maximum number of bytes that each import thread reads per round in a job, in bytes. The range is 100 MB to 1 GB, and the default value is 1 GB.

Glossary:

• Offset: The logical position of each message in a Kafka partition. OFFSET_BEGINNING indicates starting from the earliest readable message in the partition, while OFFSET_END indicates starting from the current latest position.
• low / high watermark: The minimum and maximum offsets available for consumption in a Kafka partition. The specified offset must not be lower than the low watermark.

format_options

Currently, only the CSV format is supported. Its syntax aligns with OceanBase's existing CSV parsing capabilities. For more information, see the format description in LOAD DATA. Common options are as follows:

• TYPE: The data format. Currently, only CSV is supported.
• LINE_DELIMITER: The line delimiter for the CSV file. Default value: LINE_DELIMITER='\n'.
• FIELD_DELIMITER: Optional. The column delimiter of the CSV file. Default value: FIELD_DELIMITER='\t'.
• ESCAPE: specifies the escape character for the CSV file. It must be 1 byte and defaults to ESCAPE ='\'.
• FIELD_OPTIONALLY_ENCLOSED_BY: Optional. Specifies the character that encloses field values in a CSV file. The default is an empty string. When this option is used, enclosing characters are added only to fields of certain types, such as CHAR, VARCHAR, TEXT, and JSON.
• ENCODING: Specifies the character set encoding format of the file. If not specified, the default value is UTF8MB4.
• NULL_IF: The string that will be treated as NULL. The default value is empty.
• SKIP_BLANK_LINES: Specifies whether to skip blank lines. The default value is FALSE, which means blank lines are not skipped.
• TRIM_SPACE: Specifies whether to remove leading and trailing spaces from fields in the file. The default value is FALSE, which means no spaces will be removed.
• EMPTY_FIELD_AS_NULL: Specifies whether to treat empty strings as NULL. The default value is FALSE, which means empty strings are not treated as NULL.

Examples

Create a routine load job named load_job_0 to import data from Kafka to the table test_db.test_tbl1.

obclient> CREATE ROUTINE LOAD load_job_0 ON test_db.test_tbl1
    PARALLEL = 3
    WHERE id > 0
    FROM STREAM (
        TYPE = 'KAFKA',
        'bootstrap.servers' = 'xxx.xxx.xxx.1:xxxx',
        'security.protocol' = 'sasl_plaintext',
        'sasl.mechanism' = 'PLAIN',
        'sasl.username' = '******',
        'sasl.password' = '******',
        'kafka_topic' = 'test-topic',
        'kafka_partitions' = '0,1,2',
        'kafka_offsets' = '0,0,OFFSET_BEGINNING',
        'max_batch_interval' = '3000000',
        'max_batch_rows' = '200000',
        'max_batch_size' = '104857600',
        FORMAT = (
            TYPE = 'CSV',
            FIELD_DELIMITER = ',',
            TRIM_SPACE = TRUE)
        );

References

• PAUSE ROUTINE LOAD
• RESUME ROUTINE LOAD
• STOP ROUTINE LOAD