ROUTINE LOAD

description

  1. 例行导入（Routine Load）功能，支持用户提交一个常驻的导入任务，通过不断的从指定的数据源读取数据，将数据导入到 Doris 中。
  2. 目前仅支持通过无认证或者 SSL 认证方式，从 Kakfa 导入文本格式（CSV）的数据。

语法：

  1. CREATE ROUTINE LOAD [db.]job_name ON tbl_name
  2. [load_properties]
  3. [job_properties]
  4. FROM data_source
  5. [data_source_properties]
  7. 1. [db.]job_name
  9.     导入作业的名称，在同一个 database 内，相同名称只能有一个 job 在运行。
  11. 2. tbl_name
  13.     指定需要导入的表的名称。
  15. 3. load_properties
  17.     用于描述导入数据。语法：
  19.     [column_separator],
  20.     [columns_mapping],
  21.     [where_predicates],
  22.     [partitions]
  24.     1. column_separator:
  26.         指定列分隔符，如：
  28.             COLUMNS TERMINATED BY ","
  30.         默认为：\t
  32.     2. columns_mapping:
  34.         指定源数据中列的映射关系，以及定义衍生列的生成方式。
  36.         1. 映射列：
  38.             按顺序指定，源数据中各个列，对应目的表中的哪些列。对于希望跳过的列，可以指定一个不存在的列名。
  39.             假设目的表有三列 k1, k2, v1。源数据有4列，其中第1、2、4列分别对应 k2, k1, v1。则书写如下：
  41.             COLUMNS (k2, k1, xxx, v1)
  43.             其中 xxx 为不存在的一列，用于跳过源数据中的第三列。
  45.         2. 衍生列：
  47.             以 col_name = expr 的形式表示的列，我们称为衍生列。即支持通过 expr 计算得出目的表中对应列的值。
  48.             衍生列通常排列在映射列之后，虽然这不是强制的规定，但是 Doris 总是先解析映射列，再解析衍生列。
  49.             接上一个示例，假设目的表还有第4列 v2，v2 由 k1 和 k2 的和产生。则可以书写如下：
  51.             COLUMNS (k2, k1, xxx, v1, v2 = k1 + k2);
  53.     3. where_predicates
  55.         用于指定过滤条件，以过滤掉不需要的列。过滤列可以是映射列或衍生列。
  56.         例如我们只希望导入 k1 大于 100 并且 k2 等于 1000 的列，则书写如下：
  58.         WHERE k1 > 100 and k2 = 1000
  60.     4. partitions
  62.         指定导入目的表的哪些 partition 中。如果不指定，则会自动导入到对应的 partition 中。
  63.         示例：
  65.         PARTITION(p1, p2, p3)
  67. 4. job_properties    
  69.     用于指定例行导入作业的通用参数。
  70.     语法：
  72.     PROPERTIES (
  73.         "key1" = "val1",
  74.         "key2" = "val2"
  75.     )
  77.     目前我们支持以下参数：
  79.     1. desired_concurrent_number
  81.         期望的并发度。一个例行导入作业会被分成多个子任务执行。这个参数指定一个作业最多有多少任务可以同时执行。必须大于0。默认为3。
  82.         这个并发度并不是实际的并发度，实际的并发度，会通过集群的节点数、负载情况，以及数据源的情况综合考虑。
  83.         例：
  85.             "desired_concurrent_number" = "3"
  87.     2. max_batch_interval/max_batch_rows/max_batch_size
  89.         这三个参数分别表示：
  90.         1）每个子任务最大执行时间，单位是秒。范围为 5 到 60。默认为10。
  91.         2）每个子任务最多读取的行数。必须大于等于200000。默认是200000。
  92.         3）每个子任务最多读取的字节数。单位是字节，范围是 100MB 到 1GB。默认是 100MB。
  94.         这三个参数，用于控制一个子任务的执行时间和处理量。当任意一个达到阈值，则任务结束。
  95.         例：
  97.             "max_batch_interval" = "20",
  98.             "max_batch_rows" = "300000",
  99.             "max_batch_size" = "209715200"
  101.     3. max_error_number
  103.         采样窗口内，允许的最大错误行数。必须大于等于0。默认是 0，即不允许有错误行。
  104.         采样窗口为 max_batch_rows * 10。即如果在采样窗口内，错误行数大于 max_error_number，则会导致例行作业被暂停，需要人工介入检查数据质量问题。
  105.         被 where 条件过滤掉的行不算错误行。
  107.     4. strict_mode
  109.         是否开启严格模式，默认为开启。如果开启后，非空原始数据的列类型变换如果结果为 NULL，则会被过滤。指定方式为 "strict_mode" = "true"
  111.     5. timezone
  113.         指定导入作业所使用的时区。默认为使用 Session 的 timezone 参数。该参数会影响所有导入涉及的和时区有关的函数结果。
  115. 5. data_source
  117.     数据源的类型。当前支持：
  119.         KAFKA
  121. 6. data_source_properties
  123.     指定数据源相关的信息。
  124.     语法：
  126.     (
  127.         "key1" = "val1",
  128.         "key2" = "val2"
  129.     )
  131.     1. KAFKA 数据源
  133.         1. kafka_broker_list
  135.             Kafka 的 broker 连接信息。格式为 ip:host。多个broker之间以逗号分隔。
  136.             示例：
  138.                 "kafka_broker_list" = "broker1:9092,broker2:9092"
  140.         2. kafka_topic
  142.             指定要订阅的 Kafka 的 topic。
  143.             示例：
  145.                 "kafka_topic" = "my_topic"
  147.         3. kafka_partitions/kafka_offsets
  149.             指定需要订阅的 kafka partition，以及对应的每个 partition 的起始 offset。
  151.             offset 可以指定从大于等于 0 的具体 offset，或者：
  152.             1) OFFSET_BEGINNING: 从有数据的位置开始订阅。
  153.             2) OFFSET_END: 从末尾开始订阅。
  155.             如果没有指定，则默认从 OFFSET_END 开始订阅 topic 下的所有 partition。
  156.             示例：
  158.                 "kafka_partitions" = "0,1,2,3",
  159.                 "kafka_offsets" = "101,0,OFFSET_BEGINNING,OFFSET_END" 
  161.         4. property
  163.             指定自定义kafka参数。
  164.             功能等同于kafka shell中 "--property" 参数。
  165.             当参数的 value 为一个文件时，需要在 value 前加上关键词："FILE:"。
  166.             关于如何创建文件，请参阅 "HELP CREATE FILE;"
  167.             更多支持的自定义参数，请参阅 librdkafka 的官方 CONFIGURATION 文档中，client 端的配置项。
  169.             示例:
  170.                 "property.client.id" = "12345",
  171.                 "property.ssl.ca.location" = "FILE:ca.pem"
  173.             1.使用 SSL 连接 Kafka 时，需要指定以下参数：
  175.             "property.security.protocol" = "ssl",
  176.             "property.ssl.ca.location" = "FILE:ca.pem",
  177.             "property.ssl.certificate.location" = "FILE:client.pem",
  178.             "property.ssl.key.location" = "FILE:client.key",
  179.             "property.ssl.key.password" = "abcdefg"
  181.             其中：
  182.             "property.security.protocol" 和 "property.ssl.ca.location" 为必须，用于指明连接方式为 SSL，以及 CA 证书的位置。
  184.             如果 Kafka server 端开启了 client 认证，则还需设置：
  186.             "property.ssl.certificate.location"
  187.             "property.ssl.key.location"
  188.             "property.ssl.key.password"
  190.             分别用于指定 client 的 public key，private key 以及 private key 的密码。
  193.             2.指定kafka partition的默认起始offset
  194.             如果没有指定kafka_partitions/kafka_offsets,默认消费所有分区,此时可以指定kafka_default_offsets指定起始 offset。默认为 OFFSET_END，即从末尾开始订阅。
  195.             值为
  196.                 1) OFFSET_BEGINNING: 从有数据的位置开始订阅。
  197.                 2) OFFSET_END: 从末尾开始订阅。
  198.                 示例：
  199.                 "property.kafka_default_offsets" = "OFFSET_BEGINNING"
  201. 7. 导入数据格式样例
  203.     整型类（TINYINT/SMALLINT/INT/BIGINT/LARGEINT）：1, 1000, 1234
  204.     浮点类（FLOAT/DOUBLE/DECIMAL）：1.1, 0.23, .356
  205.     日期类（DATE/DATETIME）：2017-10-03, 2017-06-13 12:34:03。
  206.     字符串类（CHAR/VARCHAR）（无引号）：I am a student, a
  207.     NULL值：\N

example

  1. 1. 为 example_db 的 example_tbl 创建一个名为 test1 的 Kafka 例行导入任务。指定group.id和client.id，并且自动默认消费所有分区，且从末尾（OFFSET_END）开始订阅
  3.     CREATE ROUTINE LOAD example_db.test1 ON example_tbl
  4.     COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100)
  5.     PROPERTIES
  6.     (
  7.         "desired_concurrent_number"="3",
  8.         "max_batch_interval" = "20",
  9.         "max_batch_rows" = "300000",
  10.         "max_batch_size" = "209715200",
  11.         "strict_mode" = "false"
  12.     )
  13.     FROM KAFKA
  14.     (
  15.         "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
  16.         "kafka_topic" = "my_topic",
  17.         "property.group.id" = "xxx",
  18.         "property.client.id" = "xxx"
  19.     );
  21. 2. 为 example_db 的 example_tbl 创建一个名为 test1 的 Kafka 例行导入任务。导入任务为严格模式。
  23.     CREATE ROUTINE LOAD example_db.test1 ON example_tbl
  24.     COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100),
  25.     WHERE k1 > 100 and k2 like "%doris%"
  26.     PROPERTIES
  27.     (
  28.         "desired_concurrent_number"="3",
  29.         "max_batch_interval" = "20",
  30.         "max_batch_rows" = "300000",
  31.         "max_batch_size" = "209715200",
  32.         "strict_mode" = "false"
  33.     )
  34.     FROM KAFKA
  35.     (
  36.         "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
  37.         "kafka_topic" = "my_topic",
  38.         "kafka_partitions" = "0,1,2,3",
  39.         "kafka_offsets" = "101,0,0,200"
  40.     );
  42. 3. 通过 SSL 认证方式，从 Kafka 集群导入数据。同时设置 client.id 参数。导入任务为非严格模式，时区为 Africa/Abidjan
  44.     CREATE ROUTINE LOAD example_db.test1 ON example_tbl
  45.     COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100),
  46.     WHERE k1 > 100 and k2 like "%doris%"
  47.     PROPERTIES
  48.     (
  49.         "desired_concurrent_number"="3",
  50.         "max_batch_interval" = "20",
  51.         "max_batch_rows" = "300000",
  52.         "max_batch_size" = "209715200",
  53.         "strict_mode" = "false",
  54.         "timezone" = "Africa/Abidjan"
  55.     )
  56.     FROM KAFKA
  57.     (
  58.         "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
  59.         "kafka_topic" = "my_topic",
  60.         "property.security.protocol" = "ssl",
  61.         "property.ssl.ca.location" = "FILE:ca.pem",
  62.         "property.ssl.certificate.location" = "FILE:client.pem",
  63.         "property.ssl.key.location" = "FILE:client.key",
  64.         "property.ssl.key.password" = "abcdefg",
  65.         "property.client.id" = "my_client_id"
  66.     );

keyword

  1. CREATE,ROUTINE,LOAD