文档中心

上云无忧 > 文档中心 > 百度数据仓库 Palo SQL语法手册 - ROUTINE-LOAD
百度智能云
百度数据仓库 Palo Doris版
简介/价格/文档
百度数据仓库 Palo SQL语法手册 - ROUTINE-LOAD

文档简介：
ROUTINE LOAD Description： 例行导入（Routine Load）功能，支持用户提交一个常驻的导入任务，通过不断的从指定的数据源读取数据，将数据导入到 PALO 中。 目前仅支持通过无认证或者 SSL 认证方式，从 Kakfa 导入 CSV 或 Json 格式的数据。
*此产品及展示信息均由百度智能云官方提供。免费试用 咨询热线：400-826-7010，为您提供专业的售前咨询，让您快速了解云产品，助您轻松上云！ 微信咨询
  免费试用、价格特惠

ROUTINE LOAD

Description

例行导入（Routine Load）功能，支持用户提交一个常驻的导入任务，通过不断的从指定的数据源读取数据，将数据导入到 PALO 中。

目前仅支持通过无认证或者 SSL 认证方式，从 Kakfa 导入 CSV 或 Json 格式的数据。

语法： 


CREATE ROUTINE LOAD [db.]job_name ON tbl_name [merge_type] [load_properties] [job_properties] 
FROM data_source [data_source_properties]


  • [db.]job_name

    导入作业的名称，在同一个 database 内，相同名称只能有一个 job 在运行。

  • tbl_name

    指定需要导入的表的名称。

  • merge_type

    数据合并类型。默认为 APPEND，表示导入的数据都是普通的追加写操作。MERGE 和 DELETE 类型仅适用于 Unique Key 模型表。其中 MERGE 类型需要配合 [DELETE ON] 语句使用，以标注 Delete Flag 列。而 DELETE 类型则表示导入的所有数据皆为删除数据。

  • load_properties

    用于描述导入数据。组成如下： 

    [column_separator],
[columns_mapping],
[preceding_filter],
[where_predicates],
[partitions],
[DELETE ON],
[ORDER BY]

    • column_separator

      指定列分隔符，默认为 \t

      COLUMNS TERMINATED BY ","

    • columns_mapping

      用于指定文件列和表中列的映射关系，以及各种列转换等。关于这部分详细介绍，可以参阅 [列的映射，转换与过滤] 文档。

      (k1, k2, tmpk1, k3 = tmpk1 + 1)

    • preceding_filter

      过滤原始数据。关于这部分详细介绍，可以参阅 [列的映射，转换与过滤] 文档。

    • where_predicates

      根据条件对导入的数据进行过滤。关于这部分详细介绍，可以参阅 [列的映射，转换与过滤] 文档。

      WHERE k1 > 100 and k2 = 1000

    • partitions

      指定导入目的表的哪些 partition 中。如果不指定，则会自动导入到对应的 partition 中。

      PARTITION(p1, p2, p3)

    • DELETE ON

      需配合 MEREGE 导入模式一起使用，仅针对 Unique Key 模型的表。用于指定导入数据中表示 Delete Flag 的列和计算关系。

      DELETE ON v3 >100

    • ORDER BY

      仅针对 Unique Key 模型的表。用于指定导入数据中表示 Sequence Col 的列。主要用于导入时保证数据顺序。

  • job_properties

    用于指定例行导入作业的通用参数。 

    PROPERTIES (
    "key1" = "val1",
    "key2" = "val2"
)

    目前我们支持以下参数：

    1. desired_concurrent_number

      期望的并发度。一个例行导入作业会被分成多个子任务执行。这个参数指定一个作业最多有多少任务可以同时执行。必须大于0。默认为3。

      这个并发度并不是实际的并发度，实际的并发度，会通过集群的节点数、负载情况，以及数据源的情况综合考虑。

      "desired_concurrent_number" = "3"

    2. max_batch_interval/max_batch_rows/max_batch_size

      这三个参数分别表示：

      1. 每个子任务最大执行时间，单位是秒。范围为 5 到 60。默认为10。
      2. 每个子任务最多读取的行数。必须大于等于200000。默认是200000。
      3. 每个子任务最多读取的字节数。单位是字节，范围是 100MB 到 1GB。默认是 100MB。

      这三个参数，用于控制一个子任务的执行时间和处理量。当任意一个达到阈值，则任务结束。 

      "max_batch_interval" = "20",
"max_batch_rows" = "300000",
"max_batch_size" = "209715200"

    3. max_error_number

      采样窗口内，允许的最大错误行数。必须大于等于0。默认是 0，即不允许有错误行。

      采样窗口为 max_batch_rows * 10。即如果在采样窗口内，错误行数大于 max_error_number，则会导致例行作业被暂停，需要人工介入检查数据质量问题。

      被 where 条件过滤掉的行不算错误行。

    4. strict_mode

      是否开启严格模式，默认为关闭。如果开启后，非空原始数据的列类型变换如果结果为 NULL，则会被过滤。指定方式为：

      "strict_mode" = "true"

    5. timezone

      指定导入作业所使用的时区。默认为使用 Session 的 timezone 参数。该参数会影响所有导入涉及的和时区有关的函数结果。

    6. format

      指定导入数据格式，默认是csv，支持json格式。

    7. jsonpaths

      当导入数据格式为 json 时，可以通过 jsonpaths 指定抽取 Json 数据中的字段。

      -H "jsonpaths: [\"$.k2\", \"$.k1\"]"

    8. strip_outer_array

      当导入数据格式为 json 时，strip_outer_array 为 true 表示 Json 数据以数组的形式展现，数据中的每一个元素将被视为一行数据。默认值是 false。

      -H "strip_outer_array: true"

    9. json_root

      当导入数据格式为 json 时，可以通过 json_root 指定 Json 数据的根节点。PALO 将通过 json_root 抽取根节点的元素进行解析。默认为空。

      -H "json_root: $.RECORDS"

  • FROM data_source [data_source_properties]

    数据源的类型。当前支持： 

    FROM KAFKA
(
    "key1" = "val1",
    "key2" = "val2"
)

    data_source_properties 支持如下数据源属性：

    1. kafka_broker_list

      Kafka 的 broker 连接信息。格式为 ip:host。多个broker之间以逗号分隔。

      "kafka_broker_list" = "broker1:9092,broker2:9092"

    2. kafka_topic

      指定要订阅的 Kafka 的 topic。

      "kafka_topic" = "my_topic"

    3. kafka_partitions/kafka_offsets

      指定需要订阅的 kafka partition，以及对应的每个 partition 的起始 offset。如果指定时间，则会从大于等于该时间的最近一个 offset 处开始消费。

      offset 可以指定从大于等于 0 的具体 offset，或者：

      • OFFSET_BEGINNING: 从有数据的位置开始订阅。
      • OFFSET_END: 从末尾开始订阅。
      • 时间格式，如："2021-05-22 11:00:00"

      如果没有指定，则默认从 OFFSET_END 开始订阅 topic 下的所有 partition。 

      "kafka_partitions" = "0,1,2,3",
"kafka_offsets" = "101,0,OFFSET_BEGINNING,OFFSET_END"
      "kafka_partitions" = "0,1,2,3",
"kafka_offsets" = "2021-05-22 11:00:00,2021-05-22 11:00:00,2021-05-22 11:00:00"

      注意，时间格式不能和 OFFSET 格式混用。

    4. property

      指定自定义kafka参数。功能等同于kafka shell中 "--property" 参数。

      当参数的 value 为一个文件时，需要在 value 前加上关键词："FILE:"。

      关于如何创建文件，请参阅 CREATE FILE 命令文档。

      更多支持的自定义参数，请参阅 librdkafka 的官方 CONFIGURATION 文档中，client 端的配置项。如： 

      "property.client.id" = "12345",
"property.ssl.ca.location" = "FILE:ca.pem"

      1. 使用 SSL 连接 Kafka 时，需要指定以下参数： 

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

        其中：

        property.security.protocol 和 property.ssl.ca.location 为必须，用于指明连接方式为 SSL，以及 CA 证书的位置。

        如果 Kafka server 端开启了 client 认证，则还需设置： 

        "property.ssl.certificate.location"
"property.ssl.key.location"
"property.ssl.key.password"

        分别用于指定 client 的 public key，private key 以及 private key 的密码。

      2. 指定kafka partition的默认起始offset

        如果没有指定 kafka_partitions/kafka_offsets，默认消费所有分区。

        此时可以指定 kafka_default_offsets 指定起始 offset。默认为 OFFSET_END，即从末尾开始订阅。

        示例： 

        "property.kafka_default_offsets" = "OFFSET_BEGINNING"

Example

  1. 为 example_db 的 example_tbl 创建一个名为 test1 的 Kafka 例行导入任务。指定列分隔符和 group.id 和 client.id，并且自动默认消费所有分区，且从有数据的位置（OFFSET_BEGINNING）开始订阅 

    CREATE ROUTINE LOAD example_db.test1 ON example_tbl COLUMNS TERMINATED BY ",", 
  2. COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100) PROPERTIES ( "desired_concurrent_number"="3", 
  3. "max_batch_interval" = "20", "max_batch_rows" = "300000", "max_batch_size" = "209715200",
  4.  "strict_mode" = "false" ) FROM KAFKA ( "kafka_broker_list" = "broker1:9092,broker2:9092,
  5. broker3:9092", "kafka_topic" = "my_topic", "property.group.id" = "xxx", "property.client.id
  6. " = "xxx", "property.kafka_default_offsets" = "OFFSET_BEGINNING" );

  7. 为 example_db 的 example_tbl 创建一个名为 test1 的 Kafka 例行导入任务。导入任务为严格模式。 

    CREATE ROUTINE LOAD example_db.test1 ON example_tbl COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100), 
  8. PRECEDING FILTER k1 = 1, WHERE k1 > 100 and k2 like "%doris%" PROPERTIES ( "desired_concurrent
  9. _number"="3", "max_batch_interval" = "20", "max_batch_rows" = "300000", "max_batch_size" = "
  10. 209715200", "strict_mode" = "false" ) FROM KAFKA ( "kafka_broker_list" = "broker1:9092,broker2:9092,
  11. broker3:9092", "kafka_topic" = "my_topic", "kafka_partitions" = "0,1,2,3", "kafka_offsets" = "101,0,0,200" );

  12. 通过 SSL 认证方式，从 Kafka 集群导入数据。同时设置 client.id 参数。导入任务为非严格模式，时区为 Africa/Abidjan 

    CREATE ROUTINE LOAD example_db.test1 ON example_tbl COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100), WHERE k1 
  13. > 100 and k2 like "%doris%" PROPERTIES ( "desired_concurrent_number"="3", "max_batch_interval" = "20",
  14.  "max_batch_rows" = "300000", "max_batch_size" = "209715200", "strict_mode" = "false", "timezone" = "
  15. Africa/Abidjan" ) FROM KAFKA ( "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
  16.  "kafka_topic" = "my_topic", "property.security.protocol" = "ssl", "property.ssl.ca.location" = "
  17. FILE:ca.pem", "property.ssl.certificate.location" = "FILE:client.pem", "property.ssl.key.location
  18. " = "FILE:client.key", "property.ssl.key.password" = "abcdefg", "property.client.id" = "my_client_id" );

  19. 导入 Json 格式数据。默认使用 Json 中的字段名作为列名映射。指定导入 0,1,2 三个分区，起始 offset 都为 0 

    CREATE ROUTINE LOAD example_db.test_json_label_1 ON table1 COLUMNS(category,price,author) PROPERTIES 
  20. ( "desired_concurrent_number"="3", "max_batch_interval" = "20", "max_batch_rows" = "300000", 
  21. "max_batch_size" = "209715200", "strict_mode" = "false", "format" = "json" ) FROM KAFKA ( "kafka_broker
  22. _list" = "broker1:9092,broker2:9092,broker3:9092", "kafka_topic" = "my_topic", "kafka_partitions" = "
  23. 0,1,2", "kafka_offsets" = "0,0,0" );

  24. 导入 Json 数据，并通过 Jsonpaths 抽取字段，并指定 Json 文档根节点 

    CREATE ROUTINE LOAD example_db.test1 ON example_tbl COLUMNS(category, author, price, timestamp,
  25.  dt=from_unixtime(timestamp, '%Y%m%d')) PROPERTIES ( "desired_concurrent_number"="3", "
  26. max_batch_interval" = "20", "max_batch_rows" = "300000", "max_batch_size" = "209715200", "
  27. strict_mode" = "false", "format" = "json", "jsonpaths" = "[\"$.category\",\"$.author\",\"$.price\",
  28. \"$.timestamp\"]", "json_root" = "$.RECORDS" "strip_outer_array" = "true" ) FROM KAFKA ( "kafka_broker
  29. _list" = "broker1:9092,broker2:9092,broker3:9092", "kafka_topic" = "my_topic", "kafka_partitions" = "
  30. 0,1,2", "kafka_offsets" = "0,0,0" );

  31. 为 example_db 的 example_tbl 创建一个名为 test1 的 Kafka 例行导入任务。并且使用条件过滤。 

    CREATE ROUTINE LOAD example_db.test1 ON example_tbl WITH MERGE COLUMNS(k1, k2, k3, v1, v2, v3), WHERE k1 >
  32.  100 and k2 like "%doris%", DELETE ON v3 >100 PROPERTIES ( "desired_concurrent_number"="3", "max_batch
  33. _interval" = "20", "max_batch_rows" = "300000", "max_batch_size" = "209715200", "strict_mode" = "false" ) 
  34. FROM KAFKA ( "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092", "kafka_topic" = "my_topic", 
  35. "kafka_partitions" = "0,1,2,3", "kafka_offsets" = "101,0,0,200" );

  36. 导入数据到含有 sequence 列的 Unique Key 模型表中 

    CREATE ROUTINE LOAD example_db.test_job ON example_tbl COLUMNS TERMINATED BY ",",
  37.  COLUMNS(k1,k2,source_sequence,v1,v2), ORDER BY source_sequence
PROPERTIES ( "desired_concurrent_number"="3", "max_batch_interval" = "30", "max_batch_rows" = "
  38. 300000", "max_batch_size" = "209715200" ) FROM KAFKA ( "kafka_broker_list" = "broker1:9092,broker2:9092,
  39. broker3:9092", "kafka_topic" = "my_topic", "kafka_partitions" = "0,1,2,3", "kafka_offsets" = "101,0,0,200" );

  40. 从指定的时间点开始消费 

    CREATE ROUTINE LOAD example_db.test_job ON example_tbl
PROPERTIES ( "desired_concurrent_number"="3", "max_batch_interval" = "30", "max_batch_rows" = "300000", "
  41. max_batch_size" = "209715200" ) FROM KAFKA ( "kafka_broker_list" = "broker1:9092,broker2:9092", "kafka_topic
  42. " = "my_topic", "kafka_default_offset" = "2021-05-21 10:00:00" );

Keywords 

CREATE, ROUTINE, LOAD

最佳实践

  1. 关于指定消费的 Partition 和 Offset

    PALO 支持指定 Partition 和 Offset 开始消费，还支持了指定时间点进行消费的功能。这里说明下对应参数的配置关系。

    有三个相关参数：

    • kafka_partitions：指定待消费的 partition 列表，如："0, 1, 2, 3"。
    • kafka_offsets：指定每个分区的起始offset，必须和 kafka_partitions 列表个数对应。如："1000, 1000, 2000, 2000"
    • property.kafka_default_offset：指定分区默认的起始offset。

    在创建导入作业时，这三个参数可以有以下组合：

    组合 kafka_partitions kafka_offsets property.kafka_default_offset 行为
    1 No No No 系统会自动查找topic对应的所有分区并从 OFFSET_END 开始消费
    2 No No Yes 系统会自动查找topic对应的所有分区并从 default offset 指定的位置开始消费
    3 Yes No No 系统会从指定分区的 OFFSET_END 开始消费
    4 Yes Yes No 系统会从指定分区的指定offset 处开始消费
    5 Yes No Yes 系统会从指定分区，default offset 指定的位置开始消费
相似文档
  • 百度数据仓库 Palo SQL语法手册 - SELECT-INTO-OUTFILE
    SELECT INTO OUTFILE Description： 该命令用于将 SQL 产出的结果集导出到本地磁盘，或通过 Broker 导出到远端存储上。 query_stmt INTO OUTFILE "file_path" [format_as] [properties]
  • 百度数据仓库 Palo SQL语法手册 - STREAM-LOAD
    STREAM LOAD Description： Stream Load 通过 HTTP 协议向 PALO 传输并导入数据。该方式主要用户本地数据的上传的导入。但其本质上是一种导入框架，其提供的 HTTP 接口，不仅能够支持本地数据的传输，也能够支持内存数据、管道数据等向 HTTP 端口传输数据。
  • 百度数据仓库 Palo SQL语法手册 - SHOW-ALTER-TABLE-MATERIALIZED-VIEW
    SHOW ALTER TABLE MATERIALIZED VIEW Description： 该命令用于查看通过 CREATE-MATERIALIZED-VIEW 语句提交的创建物化视图作业的执行情况。 该语句等同于 SHOW ALTER TABLE ROLLUP;
  • 百度数据仓库 Palo SQL语法手册 - SHOW-BACKUP
    SHOW BACKUP Description： 该语句用于查看 BACKUP 任务 SHOW BACKUP [FROM db_name]; 返回结果说明： mysql> show backup\G *************************** 1. row *************************** JobId: 89205228 SnapshotName: snapshot1
  • 百度数据仓库 Palo SQL语法手册 - SHOW-EXPORT
    SHOW EXPORT Description： 该语句用于展示指定的导出任务的执行情况。 SHOW EXPORT [FROM db_name] [ WHERE [EXPORT_JOB_ID = your_job_id] [STATE = ["PENDING"|"EXPORTING"|"FINISHED"|"CANCELLED"]] ] [ORDER BY ...] [LIMIT limit];
官方微信
联系客服
400-826-7010
7x24小时客服热线
分享
  • QQ好友
  • QQ空间
  • 微信
  • 微博
返回顶部