BROKER LOAD

description

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

语法:

  1. LOAD LABEL load_label
  2. (
  3. data_desc1[, data_desc2, ...]
  4. )
  5. WITH BROKER broker_name
  6. [broker_properties]
  7. [opt_properties];
  8. 1. load_label
  9. 当前导入批次的标签。在一个 database 内唯一。
  10. 语法:
  11. [database_name.]your_label
  12. 2. data_desc
  13. 用于描述一批导入数据。
  14. 语法:
  15. DATA INFILE
  16. (
  17. "file_path1"[, file_path2, ...]
  18. )
  19. [NEGATIVE]
  20. INTO TABLE `table_name`
  21. [PARTITION (p1, p2)]
  22. [COLUMNS TERMINATED BY "column_separator"]
  23. [FORMAT AS "file_type"]
  24. [(column_list)]
  25. [SET (k1 = func(k2))]
  26. [WHERE predicate]
  27. 说明:
  28. file_path:
  29. 文件路径,可以指定到一个文件,也可以用 * 通配符指定某个目录下的所有文件。通配符必须匹配到文件,而不能是目录。
  30. PARTITION:
  31. 如果指定此参数,则只会导入指定的分区,导入分区以外的数据会被过滤掉。
  32. 如果不指定,默认导入table的所有分区。
  33. NEGATIVE
  34. 如果指定此参数,则相当于导入一批“负”数据。用于抵消之前导入的同一批数据。
  35. 该参数仅适用于存在 value 列,并且 value 列的聚合类型仅为 SUM 的情况。
  36. column_separator
  37. 用于指定导入文件中的列分隔符。默认为 \t
  38. 如果是不可见字符,则需要加\\x作为前缀,使用十六进制来表示分隔符。
  39. hive文件的分隔符\x01,指定为"\\x01"
  40. file_type
  41. 用于指定导入文件的类型,例如:parquetorccsv。默认值通过文件后缀名判断。
  42. column_list
  43. 用于指定导入文件中的列和 table 中的列的对应关系。
  44. 当需要跳过导入文件中的某一列时,将该列指定为 table 中不存在的列名即可。
  45. 语法:
  46. (col_name1, col_name2, ...)
  47. SET:
  48. 如果指定此参数,可以将源文件某一列按照函数进行转化,然后将转化后的结果导入到table中。语法为 `column_name` = expression。举几个例子帮助理解。
  49. 1: 表中有3个列“c1, c2, c3", 源文件中前两列依次对应(c1,c2),后两列之和对应c3;那么需要指定 columns (c1,c2,tmp_c3,tmp_c4) SET (c3=tmp_c3+tmp_c4);
  50. 例2: 表中有3个列“year, month, day"三个列,源文件中只有一个时间列,为”2018-06-01 01:02:03“格式。
  51. 那么可以指定 columns(tmp_time) set (year = year(tmp_time), month=month(tmp_time), day=day(tmp_time)) 完成导入。
  52. WHERE:
  53. 对做完 transform 的数据进行过滤,符合 where 条件的数据才能被导入。WHERE 语句中只可引用表中列名。
  54. 3. broker_name
  55. 所使用的 broker 名称,可以通过 show broker 命令查看。
  56. 4. broker_properties
  57. 用于提供通过 broker 访问数据源的信息。不同的 broker,以及不同的访问方式,需要提供的信息不同。
  58. 1. Baidu HDFS/AFS
  59. 访问百度内部的 hdfs/afs 目前仅支持简单认证,需提供:
  60. usernamehdfs 用户名
  61. passwordhdfs 密码
  62. 2. BOS
  63. 需提供:
  64. bos_endpointBOS endpoint
  65. bos_accesskey:公有云用户的 accesskey
  66. bos_secret_accesskey:公有云用户的 secret_accesskey
  67. 3. Apache HDFS
  68. 社区版本的 hdfs,支持简单认证、kerberos 认证。以及支持 HA 配置。
  69. 简单认证:
  70. hadoop.security.authentication = simple (默认)
  71. usernamehdfs 用户名
  72. passwordhdfs 密码
  73. kerberos 认证:
  74. hadoop.security.authentication = kerberos
  75. kerberos_principal:指定 kerberos principal
  76. kerberos_keytab:指定 kerberos keytab 文件路径。该文件必须为 broker 进程所在服务器上的文件。
  77. kerberos_keytab_content:指定 kerberos keytab 文件内容经过 base64 编码之后的内容。这个跟 kerberos_keytab 配置二选一就可以。
  78. namenode HA
  79. 通过配置 namenode HA,可以在 namenode 切换时,自动识别到新的 namenode
  80. dfs.nameservices: 指定 hdfs 服务的名字,自定义,如:"dfs.nameservices" = "my_ha"
  81. dfs.ha.namenodes.xxx:自定义 namenode 的名字,多个名字以逗号分隔。其中 xxx dfs.nameservices 中自定义的名字,如 "dfs.ha.namenodes.my_ha" = "my_nn"
  82. dfs.namenode.rpc-address.xxx.nn:指定 namenode rpc地址信息。其中 nn 表示 dfs.ha.namenodes.xxx 中配置的 namenode 的名字,如:"dfs.namenode.rpc-address.my_ha.my_nn" = "host:port"
  83. dfs.client.failover.proxy.provider:指定 client 连接 namenode provider,默认为:org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider
  84. 4. Amazon S3
  85. 需提供:
  86. fs.s3a.access.keyAmazonS3access key
  87. fs.s3a.secret.keyAmazonS3secret key
  88. fs.s3a.endpointAmazonS3endpoint
  89. 4. opt_properties
  90. 用于指定一些特殊参数。
  91. 语法:
  92. [PROPERTIES ("key"="value", ...)]
  93. 可以指定如下参数:
  94. timeout 指定导入操作的超时时间。默认超时为4小时。单位秒。
  95. max_filter_ratio:最大容忍可过滤(数据不规范等原因)的数据比例。默认零容忍。
  96. exec_mem_limit 导入内存限制。默认为 2GB。单位为字节。
  97. strict mode 是否对数据进行严格限制。默认为 false
  98. timezone: 指定某些受时区影响的函数的时区,如 strftime/alignment_timestamp/from_unixtime 等等,具体请查阅 [时区] 文档。如果不指定,则使用 "Asia/Shanghai" 时区。
  99. 5. 导入数据格式样例
  100. 整型类(TINYINT/SMALLINT/INT/BIGINT/LARGEINT):1, 1000, 1234
  101. 浮点类(FLOAT/DOUBLE/DECIMAL):1.1, 0.23, .356
  102. 日期类(DATE/DATETIME):2017-10-03, 2017-06-13 12:34:03
  103. (注:如果是其他日期格式,可以在导入命令中,使用 strftime 或者 time_format 函数进行转换)
  104. 字符串类(CHAR/VARCHAR):"I am a student", "a"
  105. NULL值:\N

example

  1. 1. HDFS 导入一批数据,指定超时时间和过滤比例。使用铭文 my_hdfs_broker broker。简单认证。
  2. LOAD LABEL example_db.label1
  3. (
  4. DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  5. INTO TABLE `my_table`
  6. )
  7. WITH BROKER my_hdfs_broker
  8. (
  9. "username" = "hdfs_user",
  10. "password" = "hdfs_passwd"
  11. )
  12. PROPERTIES
  13. (
  14. "timeout" = "3600",
  15. "max_filter_ratio" = "0.1"
  16. );
  17. 其中 hdfs_host namenode hosthdfs_port fs.defaultFS 端口(默认9000
  18. 2. AFS 一批数据,包含多个文件。导入不同的 table,指定分隔符,指定列对应关系。
  19. LOAD LABEL example_db.label2
  20. (
  21. DATA INFILE("afs://afs_host:hdfs_port/user/palo/data/input/file1")
  22. INTO TABLE `my_table_1`
  23. COLUMNS TERMINATED BY ","
  24. (k1, k3, k2, v1, v2),
  25. DATA INFILE("afs://afs_host:hdfs_port/user/palo/data/input/file2")
  26. INTO TABLE `my_table_2`
  27. COLUMNS TERMINATED BY "\t"
  28. (k1, k2, k3, v2, v1)
  29. )
  30. WITH BROKER my_afs_broker
  31. (
  32. "username" = "afs_user",
  33. "password" = "afs_passwd"
  34. )
  35. PROPERTIES
  36. (
  37. "timeout" = "3600",
  38. "max_filter_ratio" = "0.1"
  39. );
  40. 3. HDFS 导入一批数据,指定hive的默认分隔符\x01,并使用通配符*指定目录下的所有文件。
  41. 使用简单认证,同时配置 namenode HA
  42. LOAD LABEL example_db.label3
  43. (
  44. DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/*")
  45. INTO TABLE `my_table`
  46. COLUMNS TERMINATED BY "\\x01"
  47. )
  48. WITH BROKER my_hdfs_broker
  49. (
  50. "username" = "hdfs_user",
  51. "password" = "hdfs_passwd",
  52. "dfs.nameservices" = "my_ha",
  53. "dfs.ha.namenodes.my_ha" = "my_namenode1, my_namenode2",
  54. "dfs.namenode.rpc-address.my_ha.my_namenode1" = "nn1_host:rpc_port",
  55. "dfs.namenode.rpc-address.my_ha.my_namenode2" = "nn2_host:rpc_port",
  56. "dfs.client.failover.proxy.provider" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
  57. )
  58. 4. HDFS 导入一批“负”数据。同时使用 kerberos 认证方式。提供 keytab 文件路径。
  59. LOAD LABEL example_db.label4
  60. (
  61. DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/old_file)
  62. NEGATIVE
  63. INTO TABLE `my_table`
  64. COLUMNS TERMINATED BY "\t"
  65. )
  66. WITH BROKER my_hdfs_broker
  67. (
  68. "hadoop.security.authentication" = "kerberos",
  69. "kerberos_principal"="doris@YOUR.COM",
  70. "kerberos_keytab"="/home/palo/palo.keytab"
  71. )
  72. 5. 从 HDFS 导入一批数据,指定分区。同时使用 kerberos 认证方式。提供 base64 编码后的 keytab 文件内容。
  73. LOAD LABEL example_db.label5
  74. (
  75. DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  76. INTO TABLE `my_table`
  77. PARTITION (p1, p2)
  78. COLUMNS TERMINATED BY ","
  79. (k1, k3, k2, v1, v2)
  80. )
  81. WITH BROKER my_hdfs_broker
  82. (
  83. "hadoop.security.authentication"="kerberos",
  84. "kerberos_principal"="doris@YOUR.COM",
  85. "kerberos_keytab_content"="BQIAAABEAAEACUJBSURVLkNPTQAEcGFsbw"
  86. )
  87. 6. BOS 导入一批数据,指定分区, 并对导入文件的列做一些转化,如下:
  88. 表结构为:
  89. k1 varchar(20)
  90. k2 int
  91. 假设数据文件只有一行数据:
  92. Adele,1,1
  93. 数据文件中各列,对应导入语句中指定的各列:
  94. k1,tmp_k2,tmp_k3
  95. 转换如下:
  96. 1) k1: 不变换
  97. 2) k2:是 tmp_k2 tmp_k3 数据之和
  98. LOAD LABEL example_db.label6
  99. (
  100. DATA INFILE("bos://my_bucket/input/file")
  101. INTO TABLE `my_table`
  102. PARTITION (p1, p2)
  103. COLUMNS TERMINATED BY ","
  104. (k1, tmp_k2, tmp_k3)
  105. SET (
  106. k2 = tmp_k2 + tmp_k3
  107. )
  108. )
  109. WITH BROKER my_bos_broker
  110. (
  111. "bos_endpoint" = "http://bj.bcebos.com",
  112. "bos_accesskey" = "xxxxxxxxxxxxxxxxxxxxxxxxxx",
  113. "bos_secret_accesskey"="yyyyyyyyyyyyyyyyyyyy"
  114. )
  115. 7. 导入数据到含有HLL列的表,可以是表中的列或者数据里面的列
  116. 如果表中有三列分别是(id,v1,v2,v3)。其中v1v2列是hll列。导入的源文件有3列。则(column_list)中声明第一列为id,第二三列为一个临时命名的k1,k2
  117. SET中必须给表中的hll列特殊声明 hll_hash。表中的v1列等于原始数据中的hll_hash(k1)列, 表中的v3列在原始数据中并没有对应的值,使用empty_hll补充默认值。
  118. LOAD LABEL example_db.label7
  119. (
  120. DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  121. INTO TABLE `my_table`
  122. PARTITION (p1, p2)
  123. COLUMNS TERMINATED BY ","
  124. (id, k1, k2)
  125. SET (
  126. v1 = hll_hash(k1),
  127. v2 = hll_hash(k2),
  128. v3 = empty_hll()
  129. )
  130. )
  131. WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  132. LOAD LABEL example_db.label8
  133. (
  134. DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  135. INTO TABLE `my_table`
  136. PARTITION (p1, p2)
  137. COLUMNS TERMINATED BY ","
  138. (k1, k2, tmp_k3, tmp_k4, v1, v2)
  139. SET (
  140. v1 = hll_hash(tmp_k3),
  141. v2 = hll_hash(tmp_k4)
  142. )
  143. )
  144. WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  145. 8. 导入Parquet文件中数据 指定FORMAT parquet 默认是通过文件后缀判断
  146. LOAD LABEL example_db.label9
  147. (
  148. DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  149. INTO TABLE `my_table`
  150. FORMAT AS "parquet"
  151. (k1, k2, k3)
  152. )
  153. WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  154. 9. 提取文件路径中的分区字段
  155. 如果需要,则会根据表中定义的字段类型解析文件路径中的分区字段(partitioned fields),类似SparkPartition Discovery的功能
  156. LOAD LABEL example_db.label10
  157. (
  158. DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/dir/city=beijing/*/*")
  159. INTO TABLE `my_table`
  160. FORMAT AS "csv"
  161. (k1, k2, k3)
  162. COLUMNS FROM PATH AS (city, utc_date)
  163. SET (uniq_id = md5sum(k1, city))
  164. )
  165. WITH BROKER hdfs ("username"="hdfs_user", "password"="hdfs_password");
  166. hdfs://hdfs_host:hdfs_port/user/palo/data/input/dir/city=beijing目录下包括如下文件:
  167. [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, ...]
  168. 则提取文件路径的中的cityutc_date字段
  169. 10. 对待导入数据进行过滤,k1 值大于 k2 值的列才能被导入
  170. LOAD LABEL example_db.label10
  171. (
  172. DATA INFILE("hdfs://hdfs_host:hdfs_port/user/palo/data/input/file")
  173. INTO TABLE `my_table`
  174. where k1 > k2
  175. )
  176. 11. AmazonS3 导入Parquet文件中数据,指定 FORMAT parquet,默认是通过文件后缀判断:
  177. LOAD LABEL example_db.label11
  178. (
  179. DATA INFILE("s3a://my_bucket/input/file")
  180. INTO TABLE `my_table`
  181. FORMAT AS "parquet"
  182. (k1, k2, k3)
  183. )
  184. WITH BROKER my_s3a_broker
  185. (
  186. "fs.s3a.access.key" = "xxxxxxxxxxxxxxxxxxxxxxxxxx",
  187. "fs.s3a.secret.key" = "yyyyyyyyyyyyyyyyyyyy",
  188. "fs.s3a.endpoint" = "s3.amazonaws.com"
  189. )
  190. 12. 提取文件路径中的时间分区字段,并且时间包含 %3A (在 hdfs 路径中,不允许有 ':',所有 ':' 会由 %3A 替换)
  191. 假设有如下文件:
  192. /user/data/data_time=2020-02-17 00%3A00%3A00/test.txt
  193. /user/data/data_time=2020-02-18 00%3A00%3A00/test.txt
  194. 表结构为:
  195. data_time DATETIME,
  196. k2 INT,
  197. k3 INT
  198. LOAD LABEL example_db.label12
  199. (
  200. DATA INFILE("hdfs://host:port/user/data/*/test.txt")
  201. INTO TABLE `tbl12`
  202. COLUMNS TERMINATED BY ","
  203. (k2,k3)
  204. COLUMNS FROM PATH AS (data_time)
  205. SET (data_time=str_to_date(data_time, '%Y-%m-%d %H%%3A%i%%3A%s'))
  206. )
  207. WITH BROKER "hdfs" ("username"="user", "password"="pass");

keyword

  1. BROKER,LOAD