BROKER LOAD

description

  1. Broker load 通过随 Doris 集群一同部署的 broker 进行，访问对应数据源的数据，进行数据导入。
  2. 可以通过 show broker 命令查看已经部署的 broker。
  3. 目前支持以下4种数据源：
  5. 1. Baidu HDFS：百度内部的 hdfs，仅限于百度内部使用。
  6. 2. Baidu AFS：百度内部的 afs，仅限于百度内部使用。
  7. 3. Baidu Object Storage(BOS)：百度对象存储。仅限百度内部用户、公有云用户或其他可以访问 BOS 的用户使用。
  8. 4. Apache HDFS：社区版本 hdfs。
  9. 5. Amazon S3：Amazon对象存储。

语法：

  1. LOAD LABEL load_label
  2. (
  3. data_desc1[, data_desc2, ...]
  4. )
  5. WITH BROKER broker_name
  6. [broker_properties]
  7. [opt_properties];
  9. 1. load_label
  11.     当前导入批次的标签。在一个 database 内唯一。
  12.     语法：
  13.     [database_name.]your_label
  15. 2. data_desc
  17.     用于描述一批导入数据。
  18.     语法：
  19.         [MERGE|APPEND|DELETE]
  20.         DATA INFILE
  21.         (
  22.         "file_path1"[, file_path2, ...]
  23.         )
  24.         [NEGATIVE]
  25.         INTO TABLE `table_name`
  26.         [PARTITION (p1, p2)]
  27.         [COLUMNS TERMINATED BY "column_separator"]
  28.         [FORMAT AS "file_type"]
  29.         [(column_list)]
  30.         [PRECEDING FILTER predicate]
  31.         [SET (k1 = func(k2))]
  32.         [WHERE predicate]
  33.         [DELETE ON label=true]
  34.         [ORDER BY source_sequence]
  36.     说明：
  37.         file_path: 
  39.         文件路径，可以指定到一个文件，也可以用 * 通配符指定某个目录下的所有文件。通配符必须匹配到文件，而不能是目录。
  41.         PARTITION:
  43.         如果指定此参数，则只会导入指定的分区，导入分区以外的数据会被过滤掉。
  44.         如果不指定，默认导入table的所有分区。
  46.         NEGATIVE：
  47.         如果指定此参数，则相当于导入一批“负”数据。用于抵消之前导入的同一批数据。
  48.         该参数仅适用于存在 value 列，并且 value 列的聚合类型仅为 SUM 的情况。
  50.         column_separator：
  52.         用于指定导入文件中的列分隔符。默认为 \t
  53.         如果是不可见字符，则需要加\\x作为前缀，使用十六进制来表示分隔符。
  54.         如hive文件的分隔符\x01，指定为"\\x01"
  56.         file_type：
  58.         用于指定导入文件的类型，例如：parquet、orc、csv。默认值通过文件后缀名判断。 
  60.         column_list：
  62.         用于指定导入文件中的列和 table 中的列的对应关系。
  63.         当需要跳过导入文件中的某一列时，将该列指定为 table 中不存在的列名即可。
  64.         语法：
  65.         (col_name1, col_name2, ...)
  67.         PRECEDING FILTER predicate:
  69.         用于过滤原始数据。原始数据是未经列映射、转换的数据。用户可以在对转换前的数据前进行一次过滤，选取期望的数据，再进行转换。
  71.         SET:
  73.         如果指定此参数，可以将源文件某一列按照函数进行转化，然后将转化后的结果导入到table中。语法为 `column_name` = expression。举几个例子帮助理解。
  74.         例1: 表中有3个列“c1, c2, c3", 源文件中前两列依次对应(c1,c2)，后两列之和对应c3；那么需要指定 columns (c1,c2,tmp_c3,tmp_c4) SET (c3=tmp_c3+tmp_c4); 
  75.         例2: 表中有3个列“year, month, day"三个列，源文件中只有一个时间列，为”2018-06-01 01:02:03“格式。
  76.         那么可以指定 columns(tmp_time) set (year = year(tmp_time), month=month(tmp_time), day=day(tmp_time)) 完成导入。
  78.         WHERE:
  80.         对做完 transform 的数据进行过滤，符合 where 条件的数据才能被导入。WHERE 语句中只可引用表中列名。
  82.         merge_type:
  84.         数据的合并类型，一共支持三种类型APPEND、DELETE、MERGE 其中，APPEND是默认值，表示这批数据全部需要追加到现有数据中，DELETE 表示删除与这批数据key相同的所有行，MERGE 语义 需要与delete on条件联合使用，表示满足delete 条件的数据按照DELETE 语义处理其余的按照APPEND 语义处理,
  86.         delete_on_predicates:
  88.         表示删除条件，仅在 merge type 为MERGE 时有意义，语法与where 相同
  90.         ORDER BY:
  92.         只适用于UNIQUE_KEYS,相同key列下，保证value列按照source_sequence进行REPLACE, source_sequence可以是数据源中的列，也可以是表结构中的一列。
  94. 3. broker_name
  96.     所使用的 broker 名称，可以通过 show broker 命令查看。
  98. 4. broker_properties
  100.     用于提供通过 broker 访问数据源的信息。不同的 broker，以及不同的访问方式，需要提供的信息不同。
  102.     1. Baidu HDFS/AFS
  104.         访问百度内部的 hdfs/afs 目前仅支持简单认证，需提供：
  105.         username：hdfs 用户名
  106.         password：hdfs 密码
  108.     2. BOS
  110.         需提供：
  111.         bos_endpoint：BOS 的endpoint
  112.         bos_accesskey：公有云用户的 accesskey
  113.         bos_secret_accesskey：公有云用户的 secret_accesskey
  115.     3. Apache HDFS
  117.         社区版本的 hdfs，支持简单认证、kerberos 认证。以及支持 HA 配置。
  118.         简单认证：
  119.         hadoop.security.authentication = simple (默认)
  120.         username：hdfs 用户名
  121.         password：hdfs 密码
  123.         kerberos 认证：
  124.         hadoop.security.authentication = kerberos
  125.         kerberos_principal：指定 kerberos 的 principal
  126.         kerberos_keytab：指定 kerberos 的 keytab 文件路径。该文件必须为 broker 进程所在服务器上的文件。
  127.         kerberos_keytab_content：指定 kerberos 中 keytab 文件内容经过 base64 编码之后的内容。这个跟 kerberos_keytab 配置二选一就可以。
  129.         namenode HA：
  130.         通过配置 namenode HA，可以在 namenode 切换时，自动识别到新的 namenode
  131.         dfs.nameservices: 指定 hdfs 服务的名字，自定义，如："dfs.nameservices" = "my_ha"
  132.         dfs.ha.namenodes.xxx：自定义 namenode 的名字,多个名字以逗号分隔。其中 xxx 为 dfs.nameservices 中自定义的名字，如 "dfs.ha.namenodes.my_ha" = "my_nn"
  133.         dfs.namenode.rpc-address.xxx.nn：指定 namenode 的rpc地址信息。其中 nn 表示 dfs.ha.namenodes.xxx 中配置的 namenode 的名字，如："dfs.namenode.rpc-address.my_ha.my_nn" = "host:port"
  134.         dfs.client.failover.proxy.provider：指定 client 连接 namenode 的 provider，默认为：org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider
  136.     4. Amazon S3
  138.         需提供：
  139.         fs.s3a.access.key：AmazonS3的access key
  140.         fs.s3a.secret.key：AmazonS3的secret key
  141.         fs.s3a.endpoint：AmazonS3的endpoint 
  143. 4. opt_properties
  145.     用于指定一些特殊参数。
  146.     语法：
  147.     [PROPERTIES ("key"="value", ...)]
  149.     可以指定如下参数：
  150.     timeout：         指定导入操作的超时时间。默认超时为4小时。单位秒。
  151.     max_filter_ratio：最大容忍可过滤（数据不规范等原因）的数据比例。默认零容忍。
  152.     exec_mem_limit：  导入内存限制。默认为 2GB。单位为字节。
  153.     strict mode：     是否对数据进行严格限制。默认为 false。
  154.     timezone:         指定某些受时区影响的函数的时区，如 strftime/alignment_timestamp/from_unixtime 等等，具体请查阅 [时区] 文档。如果不指定，则使用 "Asia/Shanghai" 时区。
  156. 5. 导入数据格式样例
  158.     整型类（TINYINT/SMALLINT/INT/BIGINT/LARGEINT）：1, 1000, 1234
  159.     浮点类（FLOAT/DOUBLE/DECIMAL）：1.1, 0.23, .356
  160.     日期类（DATE/DATETIME）：2017-10-03, 2017-06-13 12:34:03。
  161.     （注：如果是其他日期格式，可以在导入命令中，使用 strftime 或者 time_format 函数进行转换）
  162.     字符串类（CHAR/VARCHAR）："I am a student", "a"
  163.     NULL值：\N

example

  1. 1. 从 HDFS 导入一批数据，指定超时时间和过滤比例。使用明文 my_hdfs_broker 的 broker。简单认证。
  3.     LOAD LABEL example_db.label1
  4.     (
  5.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  6.     INTO TABLE `my_table`
  7.     )
  8.     WITH BROKER my_hdfs_broker
  9.     (
  10.     "username" = "hdfs_user",
  11.     "password" = "hdfs_passwd"
  12.     )
  13.     PROPERTIES
  14.     (
  15.     "timeout" = "3600",
  16.     "max_filter_ratio" = "0.1"
  17.     );
  19.     其中 hdfs_host 为 namenode 的 host，hdfs_port 为 fs.defaultFS 端口（默认9000）
  21. 2. 从 AFS 一批数据，包含多个文件。导入不同的 table，指定分隔符，指定列对应关系。
  23.     LOAD LABEL example_db.label2
  24.     (
  25.     DATA INFILE("afs://afs_host:hdfs_port/user/palo/data/input/file1")
  26.     INTO TABLE `my_table_1`
  27.     COLUMNS TERMINATED BY ","
  28.     (k1, k3, k2, v1, v2),
  29.     DATA INFILE("afs://afs_host:hdfs_port/user/palo/data/input/file2")
  30.     INTO TABLE `my_table_2`
  31.     COLUMNS TERMINATED BY "\t"
  32.     (k1, k2, k3, v2, v1)
  33.     )
  34.     WITH BROKER my_afs_broker
  35.     (
  36.     "username" = "afs_user",
  37.     "password" = "afs_passwd"
  38.     )
  39.     PROPERTIES
  40.     (
  41.     "timeout" = "3600",
  42.     "max_filter_ratio" = "0.1"
  43.     );
  46. 3. 从 HDFS 导入一批数据，指定hive的默认分隔符\x01，并使用通配符*指定目录下的所有文件。
  47.    使用简单认证，同时配置 namenode HA
  49.     LOAD LABEL example_db.label3
  50.     (
  51.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/*")
  52.     INTO TABLE `my_table`
  53.     COLUMNS TERMINATED BY "\\x01"
  54.     )
  55.     WITH BROKER my_hdfs_broker
  56.     (
  57.     "username" = "hdfs_user",
  58.     "password" = "hdfs_passwd",
  59.     "dfs.nameservices" = "my_ha",
  60.     "dfs.ha.namenodes.my_ha" = "my_namenode1, my_namenode2",
  61.     "dfs.namenode.rpc-address.my_ha.my_namenode1" = "nn1_host:rpc_port",
  62.     "dfs.namenode.rpc-address.my_ha.my_namenode2" = "nn2_host:rpc_port",
  63.     "dfs.client.failover.proxy.provider" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
  64.     )
  66. 4. 从 HDFS 导入一批“负”数据。同时使用 kerberos 认证方式。提供 keytab 文件路径。
  68.     LOAD LABEL example_db.label4
  69.     (
  70.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/old_file)
  71.     NEGATIVE
  72.     INTO TABLE `my_table`
  73.     COLUMNS TERMINATED BY "\t"
  74.     )
  75.     WITH BROKER my_hdfs_broker
  76.     (
  77.     "hadoop.security.authentication" = "kerberos",
  78.     "kerberos_principal"="doris@YOUR.COM",
  79.     "kerberos_keytab"="/home/palo/palo.keytab"
  80.     )
  82. 5. 从 HDFS 导入一批数据，指定分区。同时使用 kerberos 认证方式。提供 base64 编码后的 keytab 文件内容。
  84.     LOAD LABEL example_db.label5
  85.     (
  86.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  87.     INTO TABLE `my_table`
  88.     PARTITION (p1, p2)
  89.     COLUMNS TERMINATED BY ","
  90.     (k1, k3, k2, v1, v2)
  91.     )
  92.     WITH BROKER my_hdfs_broker
  93.     (
  94.     "hadoop.security.authentication"="kerberos",
  95.     "kerberos_principal"="doris@YOUR.COM",
  96.     "kerberos_keytab_content"="BQIAAABEAAEACUJBSURVLkNPTQAEcGFsbw"
  97.     )
  99. 6. 从 BOS 导入一批数据，指定分区, 并对导入文件的列做一些转化，如下：
  100.    表结构为：
  101.     k1 varchar(20)
  102.     k2 int
  104.     假设数据文件只有一行数据：
  106.     Adele,1,1
  108.     数据文件中各列，对应导入语句中指定的各列：
  109.     k1,tmp_k2,tmp_k3
  111.     转换如下：
  113.     1) k1: 不变换
  114.     2) k2：是 tmp_k2 和 tmp_k3 数据之和
  116.     LOAD LABEL example_db.label6
  117.     (
  118.     DATA INFILE("bos://my_bucket/input/file")
  119.     INTO TABLE `my_table`
  120.     PARTITION (p1, p2)
  121.     COLUMNS TERMINATED BY ","
  122.     (k1, tmp_k2, tmp_k3)
  123.     SET (
  124.       k2 = tmp_k2 + tmp_k3
  125.     )
  126.     )
  127.     WITH BROKER my_bos_broker
  128.     (
  129.     "bos_endpoint" = "http://bj.bcebos.com",
  130.     "bos_accesskey" = "xxxxxxxxxxxxxxxxxxxxxxxxxx",
  131.     "bos_secret_accesskey"="yyyyyyyyyyyyyyyyyyyy"
  132.     )
  134. 7. 导入数据到含有HLL列的表，可以是表中的列或者数据里面的列
  136.     如果表中有4列分别是（id,v1,v2,v3）。其中v1和v2列是hll列。导入的源文件有3列, 其中表中的第一列 = 源文件中的第一列，而表中的第二，三列为源文件中的第二，三列变换得到，表中的第四列在源文件中并不存在。
  137.     则（column_list）中声明第一列为id，第二三列为一个临时命名的k1,k2。
  138.     在SET中必须给表中的hll列特殊声明 hll_hash。表中的v1列等于原始数据中的hll_hash(k1)列, 表中的v3列在原始数据中并没有对应的值，使用empty_hll补充默认值。
  139.     LOAD LABEL example_db.label7
  140.     (
  141.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  142.     INTO TABLE `my_table`
  143.     PARTITION (p1, p2)
  144.     COLUMNS TERMINATED BY ","
  145.     (id, k1, k2)
  146.     SET (
  147.       v1 = hll_hash(k1),
  148.       v2 = hll_hash(k2),
  149.       v3 = empty_hll()
  150.     )
  151.     )
  152.     WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  154.     LOAD LABEL example_db.label8
  155.     (
  156.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  157.     INTO TABLE `my_table`
  158.     PARTITION (p1, p2)
  159.     COLUMNS TERMINATED BY ","
  160.     (k1, k2, tmp_k3, tmp_k4, v1, v2)
  161.     SET (
  162.       v1 = hll_hash(tmp_k3),
  163.       v2 = hll_hash(tmp_k4)
  164.     )
  165.     )
  166.     WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  168. 8. 导入Parquet文件中数据  指定FORMAT 为parquet， 默认是通过文件后缀判断
  170.     LOAD LABEL example_db.label9
  171.     (
  172.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  173.     INTO TABLE `my_table`
  174.     FORMAT AS "parquet"
  175.     (k1, k2, k3)
  176.     )
  177.     WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  179. 9. 提取文件路径中的分区字段
  181.     如果需要，则会根据表中定义的字段类型解析文件路径中的分区字段（partitioned fields），类似Spark中Partition Discovery的功能
  183.     LOAD LABEL example_db.label10
  184.     (
  185.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/dir/city=beijing/*/*")
  186.     INTO TABLE `my_table`
  187.     FORMAT AS "csv"
  188.     (k1, k2, k3)
  189.     COLUMNS FROM PATH AS (city, utc_date)
  190.     SET (uniq_id = md5sum(k1, city))
  191.     )
  192.     WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  194.     hdfs://hdfs_host:hdfs_port/user/palo/data/input/dir/city=beijing目录下包括如下文件：
  196.     [hdfs://hdfs_host:hdfs_port/user/palo/data/input/dir/city=beijing/utc_date=2019-06-26/0000.csv, hdfs://hdfs_host:hdfs_port/user/palo/data/input/dir/city=beijing/utc_date=2019-06-26/0001.csv, ...]
  198.     则提取文件路径的中的city和utc_date字段
  200. 10. 对待导入数据进行过滤，k1 值大于 k2 值的列才能被导入
  202.     LOAD LABEL example_db.label10
  203.     (
  204.     DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  205.     INTO TABLE `my_table`
  206.     where k1 > k2
  207.     )
  209. 11. 从 AmazonS3 导入Parquet文件中数据，指定 FORMAT 为parquet，默认是通过文件后缀判断：
  211.     LOAD LABEL example_db.label11
  212.     (
  213.     DATA INFILE("s3a://my_bucket/input/file")
  214.     INTO TABLE `my_table`
  215.     FORMAT AS "parquet"
  216.     (k1, k2, k3)
  217.     )
  218.     WITH BROKER my_s3a_broker
  219.     (
  220.     "fs.s3a.access.key" = "xxxxxxxxxxxxxxxxxxxxxxxxxx",
  221.     "fs.s3a.secret.key" = "yyyyyyyyyyyyyyyyyyyy",
  222.     "fs.s3a.endpoint" = "s3.amazonaws.com"
  223.     )
  225. 12. 提取文件路径中的时间分区字段，并且时间包含 %3A (在 hdfs 路径中，不允许有 ':'，所有 ':' 会由 %3A 替换)
  227.     假设有如下文件：
  229.     /user/data/data_time=2020-02-17 00%3A00%3A00/test.txt
  230.     /user/data/data_time=2020-02-18 00%3A00%3A00/test.txt
  232.     表结构为：
  233.     data_time DATETIME,
  234.     k2        INT,
  235.     k3        INT
  237.     LOAD LABEL example_db.label12
  238.     (
  239.      DATA INFILE("hdfs://host:port/user/data/*/test.txt") 
  240.      INTO TABLE `tbl12`
  241.      COLUMNS TERMINATED BY ","
  242.      (k2,k3)
  243.      COLUMNS FROM PATH AS (data_time)
  244.      SET (data_time=str_to_date(data_time, '%Y-%m-%d %H%%3A%i%%3A%s'))
  245.     ) 
  246.     WITH BROKER "hdfs" ("username"="user", "password"="pass");
  248. 13. 从 HDFS 导入一批数据，指定超时时间和过滤比例。使用明文 my_hdfs_broker 的 broker。简单认证。并且将原有数据中与 导入数据中v2 大于100 的列相匹配的列删除，其他列正常导入
  250.     LOAD LABEL example_db.label1
  251.     (
  252.     MERGE DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  253.     INTO TABLE `my_table`
  254.     COLUMNS TERMINATED BY "\t"
  255.     (k1, k2, k3, v2, v1)
  256.     )
  257.     DELETE ON v2 >100
  258.     WITH BROKER my_hdfs_broker
  259.     (
  260.     "username" = "hdfs_user",
  261.     "password" = "hdfs_passwd"
  262.     )
  263.     PROPERTIES
  264.     (
  265.     "timeout" = "3600",
  266.     "max_filter_ratio" = "0.1"
  267.     );
  269. 14. 导入时指定source_sequence列，保证UNIQUE_KEYS表中的替换顺序：
  270.     LOAD LABEL example_db.label_sequence
  271.     (
  272.      DATA INFILE("hdfs://host:port/user/data/*/test.txt")
  273.      INTO TABLE `tbl1`
  274.      COLUMNS TERMINATED BY ","
  275.      (k1,k2,source_sequence,v1,v2)
  276.      ORDER BY source_sequence
  277.     ) 
  278.     with BROKER "hdfs" ("username"="user", "password"="pass");
  280. 14. 先过滤原始数据，在进行列的映射、转换和过滤操作
  282.     LOAD LABEL example_db.label_filter
  283.     (
  284.      DATA INFILE("hdfs://host:port/user/data/*/test.txt")
  285.      INTO TABLE `tbl1`
  286.      COLUMNS TERMINATED BY ","
  287.      (k1,k2,v1,v2)
  288.      PRECEDING FILTER k1 > 2
  289.      SET (k1 = k1 +1)
  290.      WHERE k1 > 3
  291.     ) 
  292.     with BROKER "hdfs" ("username"="user", "password"="pass");

keyword

  1. BROKER,LOAD