ROUTINE LOAD

    2. [merge_type]
    3. [load_properties]
    4. [job_properties]
    5. FROM data_source
    6. [data_source_properties]
    8. 1. [db.]job_name
    10.     导入作业的名称，在同一个 database 内，相同名称只能有一个 job 在运行。
    12. 2. tbl_name
    14.     指定需要导入的表的名称。
    16. 3. merge_type
    17.     数据的合并类型，一共支持三种类型APPEND、DELETE、MERGE 其中，APPEND是默认值，表示这批数据全部需要追加到现有数据中，DELETE 表示删除与这批数据key相同的所有行，MERGE 语义 需要与delete on条件联合使用，表示满足delete 条件的数据按照DELETE 语义处理其余的按照APPEND 语义处理, 语法为[WITH MERGE|APPEND|DELETE]
    19. 4. load_properties
    21.     用于描述导入数据。语法：
    22.     [column_separator],
    23.     [columns_mapping],
    24.     [where_predicates],
    25.     [delete_on_predicates],
    26.     [source_sequence],
    27.     [partitions],
    28.     [preceding_predicates]
    30.     1. column_separator:
    32.         指定列分隔符，如：
    34.             COLUMNS TERMINATED BY ","
    36.         默认为：\t
    38.     2. columns_mapping:
    40.         指定源数据中列的映射关系，以及定义衍生列的生成方式。
    42.         1. 映射列：
    44.             按顺序指定，源数据中各个列，对应目的表中的哪些列。对于希望跳过的列，可以指定一个不存在的列名。
    45.             假设目的表有三列 k1, k2, v1。源数据有4列，其中第1、2、4列分别对应 k2, k1, v1。则书写如下：
    47.             COLUMNS (k2, k1, xxx, v1)
    49.             其中 xxx 为不存在的一列，用于跳过源数据中的第三列。
    51.         2. 衍生列：
    53.             以 col_name = expr 的形式表示的列，我们称为衍生列。即支持通过 expr 计算得出目的表中对应列的值。
    54.             衍生列通常排列在映射列之后，虽然这不是强制的规定，但是 Doris 总是先解析映射列，再解析衍生列。
    55.             接上一个示例，假设目的表还有第4列 v2，v2 由 k1 和 k2 的和产生。则可以书写如下：
    57.             COLUMNS (k2, k1, xxx, v1, v2 = k1 + k2);
    59.     3. where_predicates
    61.         用于指定过滤条件，以过滤掉不需要的列。过滤列可以是映射列或衍生列。
    62.         例如我们只希望导入 k1 大于 100 并且 k2 等于 1000 的列，则书写如下：
    64.         WHERE k1 > 100 and k2 = 1000
    66.     4. partitions
    68.         指定导入目的表的哪些 partition 中。如果不指定，则会自动导入到对应的 partition 中。
    69.         示例：
    71.         PARTITION(p1, p2, p3)
    73.     5. delete_on_predicates
    74.         表示删除条件，仅在 merge type 为MERGE 时有意义，语法与where 相同
    76.     6. source_sequence:
    78.         只适用于UNIQUE_KEYS,相同key列下，保证value列按照source_sequence列进行REPLACE, source_sequence可以是数据源中的列，也可以是表结构中的一列。
    80.     7. preceding_predicates
    82.         PRECEDING FILTER predicate
    84.         用于过滤原始数据。原始数据是未经列映射、转换的数据。用户可以在对转换前的数据前进行一次过滤，选取期望的数据，再进行转换。
    86. 5. job_properties
    88.     用于指定例行导入作业的通用参数。
    89.     语法：
    90.     PROPERTIES (
    91.         "key1" = "val1",
    93.     )
    95.     目前我们支持以下参数：
    97.     1. desired_concurrent_number
    99.         期望的并发度。一个例行导入作业会被分成多个子任务执行。这个参数指定一个作业最多有多少任务可以同时执行。必须大于0。默认为3。
    100.         这个并发度并不是实际的并发度，实际的并发度，会通过集群的节点数、负载情况，以及数据源的情况综合考虑。
    101.         例：
    103.             "desired_concurrent_number" = "3"
    105.     2. max_batch_interval/max_batch_rows/max_batch_size
    107.         这三个参数分别表示：
    108.         1）每个子任务最大执行时间，单位是秒。范围为 5 到 60。默认为10。
    109.         2）每个子任务最多读取的行数。必须大于等于200000。默认是200000。
    110.         3）每个子任务最多读取的字节数。单位是字节，范围是 100MB 到 1GB。默认是 100MB。
    112.         这三个参数，用于控制一个子任务的执行时间和处理量。当任意一个达到阈值，则任务结束。
    113.         例：
    115.             "max_batch_interval" = "20",
    116.             "max_batch_rows" = "300000",
    117.             "max_batch_size" = "209715200"
    119.     3. max_error_number
    121.         采样窗口内，允许的最大错误行数。必须大于等于0。默认是 0，即不允许有错误行。
    122.         采样窗口为 max_batch_rows * 10。即如果在采样窗口内，错误行数大于 max_error_number，则会导致例行作业被暂停，需要人工介入检查数据质量问题。
    123.         被 where 条件过滤掉的行不算错误行。
    125.     4. strict_mode
    127.         是否开启严格模式，默认为关闭。如果开启后，非空原始数据的列类型变换如果结果为 NULL，则会被过滤。指定方式为 "strict_mode" = "true"
    129.     5. timezone
    131.         指定导入作业所使用的时区。默认为使用 Session 的 timezone 参数。该参数会影响所有导入涉及的和时区有关的函数结果。
    133.     6. format
    135.         指定导入数据格式，默认是csv，支持json格式。
    137.     7. jsonpaths
    139.         jsonpaths: 导入json方式分为：简单模式和匹配模式。如果设置了jsonpath则为匹配模式导入，否则为简单模式导入，具体可参考示例。
    141.     8. strip_outer_array
    143.         布尔类型，为true表示json数据以数组对象开始且将数组对象中进行展平，默认值是false。
    145.     9. json_root
    147.         json_root为合法的jsonpath字符串，用于指定json document的根节点，默认值为""。
    149.     10. send_batch_parallelism
    151.         整型，用于设置发送批处理数据的并行度，如果并行度的值超过 BE 配置中的 `max_send_batch_parallelism_per_job`，那么作为协调点的 BE 将使用 `max_send_batch_parallelism_per_job` 的值。 
    153. 6. data_source
    155.     数据源的类型。当前支持：
    157.         KAFKA
    159. 7. data_source_properties
    161.     指定数据源相关的信息。
    162.     语法：
    164.     (
    165.         "key1" = "val1",
    166.         "key2" = "val2"
    167.     )
    169.     1. KAFKA 数据源
    171.         1. kafka_broker_list
    173.             Kafka 的 broker 连接信息。格式为 ip:host。多个broker之间以逗号分隔。
    174.             示例：
    176.                 "kafka_broker_list" = "broker1:9092,broker2:9092"
    178.         2. kafka_topic
    180.             指定要订阅的 Kafka 的 topic。
    181.             示例：
    182.                 "kafka_topic" = "my_topic"
    184.         3. kafka_partitions/kafka_offsets
    186.             指定需要订阅的 kafka partition，以及对应的每个 partition 的起始 offset。
    188.             offset 可以指定从大于等于 0 的具体 offset，或者：
    189.             1) OFFSET_BEGINNING: 从有数据的位置开始订阅。
    190.             2) OFFSET_END: 从末尾开始订阅。
    191.             3) 时间戳，格式必须如："2021-05-11 10:00:00"，系统会自动定位到大于等于该时间戳的第一个消息的offset。注意，时间戳格式的offset不能和数字类型混用，只能选其一。
    193.             如果没有指定，则默认从 OFFSET_END 开始订阅 topic 下的所有 partition。
    194.             示例：
    196.                 "kafka_partitions" = "0,1,2,3",
    197.                 "kafka_offsets" = "101,0,OFFSET_BEGINNING,OFFSET_END"
    199.                 "kafka_partitions" = "0,1",
    200.                 "kafka_offsets" = "2021-05-11 10:00:00, 2021-05-11 11:00:00"
    202.         4. property
    204.             指定自定义kafka参数。
    205.             功能等同于kafka shell中 "--property" 参数。
    206.             当参数的 value 为一个文件时，需要在 value 前加上关键词："FILE:"。
    207.             关于如何创建文件，请参阅 "HELP CREATE FILE;"
    208.             更多支持的自定义参数，请参阅 librdkafka 的官方 CONFIGURATION 文档中，client 端的配置项。
    210.             示例:
    211.                 "property.client.id" = "12345",
    212.                 "property.ssl.ca.location" = "FILE:ca.pem"
    214.             1.使用 SSL 连接 Kafka 时，需要指定以下参数：
    216.             "property.security.protocol" = "ssl",
    217.             "property.ssl.ca.location" = "FILE:ca.pem",
    218.             "property.ssl.certificate.location" = "FILE:client.pem",
    219.             "property.ssl.key.location" = "FILE:client.key",
    220.             "property.ssl.key.password" = "abcdefg"
    222.             其中：
    223.             "property.security.protocol" 和 "property.ssl.ca.location" 为必须，用于指明连接方式为 SSL，以及 CA 证书的位置。
    225.             如果 Kafka server 端开启了 client 认证，则还需设置：
    227.             "property.ssl.certificate.location"
    228.             "property.ssl.key.location"
    229.             "property.ssl.key.password"
    231.             分别用于指定 client 的 public key，private key 以及 private key 的密码。
    234.             2.指定kafka partition的默认起始offset
    235.             如果没有指定kafka_partitions/kafka_offsets,默认消费所有分区,此时可以指定kafka_default_offsets指定起始 offset。默认为 OFFSET_END，即从末尾开始订阅。
    236.             值为
    237.                 1) OFFSET_BEGINNING: 从有数据的位置开始订阅。
    238.                 2) OFFSET_END: 从末尾开始订阅。
    239.                 3) 时间戳，格式同 kafka_offsets
    241.                 示例：
    242.                 "property.kafka_default_offsets" = "OFFSET_BEGINNING"
    243.                 "property.kafka_default_offsets" = "2021-05-11 10:00:00"
    245. 8. 导入数据格式样例
    247.     整型类（TINYINT/SMALLINT/INT/BIGINT/LARGEINT）：1, 1000, 1234
    248.     浮点类（FLOAT/DOUBLE/DECIMAL）：1.1, 0.23, .356
    249.     日期类（DATE/DATETIME）：2017-10-03, 2017-06-13 12:34:03。
    250.     字符串类（CHAR/VARCHAR）（无引号）：I am a student, a
    251.     NULL值：\N
    1. 6. 用户指定根节点json_root
    2.     CREATE ROUTINE LOAD example_db.test1 ON example_tbl
    3.     COLUMNS(category, author, price, timestamp, dt=from_unixtime(timestamp, '%Y%m%d'))
    4.     PROPERTIES
    5.     (
    6.         "desired_concurrent_number"="3",
    7.         "max_batch_interval" = "20",
    8.         "max_batch_rows" = "300000",
    9.         "max_batch_size" = "209715200",
    10.         "strict_mode" = "false",
    11.         "format" = "json",
    12.         "jsonpaths" = "[\"$.category\",\"$.author\",\"$.price\",\"$.timestamp\"]",
    13.         "strip_outer_array" = "true",
    14.         "json_root" = "$.RECORDS"
    15.     )
    16.     FROM KAFKA
    17.     (
    18.         "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
    19.         "kafka_topic" = "my_topic",
    20.         "kafka_partitions" = "0,1,2",
    21.         "kafka_offsets" = "0,0,0"
    22.     );