CREATE ROUTINE LOAD

Purpose

This statement is used to create a routine load task that continuously consumes data from Kafka and imports it into a table 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 Oracle-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 primary key or unique key conflicts.

  • 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: The list of Kafka broker addresses. Separate multiple addresses with commas, for example, broker1:9092,broker2:9092. This parameter specifies the entry address for the client to connect to Kafka for the first time.
• kafka_topic: The name of the Kafka topic to consume. This parameter is required.
• 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 the job, in seconds. The default value is 60s (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 value 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 currently 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. Specifies the column delimiter for the CSV file. The default value is FIELD_DELIMITER='\t'.
• ESCAPE: Specifies the escape character for the CSV file. It must be one byte in length. The default value is ESCAPE ='\'.
• FIELD_OPTIONALLY_ENCLOSED_BY: Optional. Specifies the character that encloses field values in a CSV file. The default is an empty string. Use this option to indicate that enclosing characters are added only for certain types of fields (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 delete leading and trailing spaces from fields in the file. The default value is FALSE, which means that leading and trailing spaces are not deleted.
• 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