Broker Load

语法

LOAD LABEL load_label
(
data_desc1[, data_desc2, ...]
)
WITH broker_type
[broker_properties]
[load_properties];

参数说明

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

data_desc1参数说明

[MERGE|APPEND|DELETE]
DATA INFILE
(
"file_path1"[, file_path2, ...]
)
[NEGATIVE]
INTO TABLE `table_name`
[PARTITION (p1, p2, ...)]
[COLUMNS TERMINATED BY "column_separator"]
[FORMAT AS "file_type"]
[(column_list)]
[COLUMNS FROM PATH AS (c1, c2, ...)]
[PRECEDING FILTER predicate]
[SET (column_mapping)]
[WHERE predicate]
[DELETE ON expr]
[ORDER BY source_sequence]
[PROPERTIES ("key1"="value1", ...)]

load_properties参数说明

使用示例

1. 创建待导入的SelectDB数据表，示例如下。

   CREATE TABLE test_table
   (
       id int,
       name varchar(50),
       age int,
       address varchar(50)
   )
   UNIQUE KEY(`id`)
   DISTRIBUTED BY HASH(id) BUCKETS 4
   PROPERTIES("replication_num" = "1");
   
   CREATE TABLE test_table2
   (
       id int,
       name varchar(50),
       age int,
       address varchar(50)
   )
   DISTRIBUTED BY HASH(id) BUCKETS 4
   PROPERTIES("replication_num" = "1");

2. 创建待导入的文件。

   • 文件file1.txt，文件内容如下。

     1,tomori,32,shanghai
     2,anon,22,beijing
     3,taki,23,shenzhen
     4,rana,45,hangzhou
     5,soyo,14,shanghai
     6,saki,25,hangzhou
     7,mutsumi,45,shanghai
     8,uika,26,shanghai
     9,umiri,27,shenzhen
     10,nyamu,37,shanghai

   • 文件file2.csv，文件内容如下。

     1,saki,25,hangzhou
     2,mutsumi,45,shanghai
     3,uika,26,shanghai
     4,umiri,27,shenzhen
     5,nyamu,37,shanghai

3. 将文件数据导入到表中。

   • 从HDFS导入文本文件file1.txt，示例如下。

     LOAD LABEL example_db.label1
     (
         DATA INFILE("hdfs://hdfs_host:hdfs_port/example/file1.txt")
         INTO TABLE `my_table`
         COLUMNS TERMINATED BY ","
     )
     WITH HDFS
     (
         "fs.defaultFS" = "hdfs://hdfs_host:hdfs_port"
     );

     导入文件file1.txt，按逗号分隔，导入到表test_table。当从HDFS导入数据时，broker_properties中需要指定fs.defaultFS属性，以确保可以正确的连接到HDFS集群并找到相应的数据文件。

   • 从HDFS导入数据，同时导入两个文件到两个表中，示例如下。

     LOAD LABEL test_db.test_02
     (
         DATA INFILE("hdfs://hdfs_host:hdfs_port/example/file2.csv")
         INTO TABLE `test_table`
         COLUMNS TERMINATED BY ","
         (id,name,temp_age,address)    
         SET (
             age = temp_age + 1
         ),
         DATA INFILE("hdfs://hdfs_host:hdfs_port/example/file1.txt")
         INTO TABLE `test_table2`
         COLUMNS TERMINATED BY ","
     )
     WITH HDFS
     (
         "fs.defaultFS" = "hdfs://hdfs_host:hdfs_port"
     );

     导入两个文件file1.txt和file2.csv，分别导入到test_table和test_table2两张表中，并且将file2.csv中age的值加1后导入。

   • 从HA模式部署的HDFS集群中，导入一批数据，示例如下。

     LOAD LABEL test_db.test_03
     (
         DATA INFILE("hdfs://hdfs_host:hdfs_port/example/*")
         INTO TABLE `test_table`
         COLUMNS TERMINATED BY "\\x01"
     )
     WITH HDFS
     (
         "hadoop.username" = "hive",
         "fs.defaultFS" = "hdfs://my_ha",
         "dfs.nameservices" = "my_ha",
         "dfs.ha.namenodes.my_ha" = "my_namenode1, my_namenode2",
         "dfs.namenode.rpc-address.my_ha.my_namenode1" = "nn1_host:rpc_port",
         "dfs.namenode.rpc-address.my_ha.my_namenode2" = "nn2_host:rpc_port",
         "dfs.client.failover.proxy.provider.my_ha" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
     );

     指定分隔符为Hive的默认分隔符\\x01，并使用通配符*指定data目录下所有目录的所有文件。

   • 对导入数据file1.txt进行过滤处理，符合条件的数据才可导入，示例如下。

     LOAD LABEL test_db.test_04
     (
         DATA INFILE("hdfs://host:port/example/file1.txt")
         INTO TABLE `test_table2`
         COLUMNS TERMINATED BY ","
         (id,name,age,address)   
         WHERE age < 20
     )
     WITH HDFS
     (
         "fs.defaultFS" = "hdfs://hdfs_host:hdfs_port"
     );
     

     只有原始数据中age < 20的行才会被导入。

   • 从HDFS导入一批数据file1.txt，指定超时时间和过滤比例，并且将原有数据中与导入数据中age<20的列相匹配的行删除，其他行正常导入，示例如下。

     LOAD LABEL test_db.test_05
     (
         MERGE DATA INFILE("hdfs://hdfs_host:hdfs_port/example/file1.txt")
         INTO TABLE `test_table`
         COLUMNS TERMINATED BY ","
         (id,name,age,address)   
         DELETE ON age < 20
     )
     WITH HDFS
     (
         "fs.defaultFS" = "hdfs://hdfs_host:hdfs_port"
     )
     PROPERTIES
     (
         "timeout" = "3600",
         "max_filter_ratio" = "0.1"
     );

     使用MERGE方式导入。test_table必须是一张Unique Key的表。当导入数据中的age列的值小于20时，该行会被认为是一个删除行。导入任务的超时时间是3600秒，并且允许错误率在10%以内。

语法

CANCEL LOAD
[FROM db_name]
WHERE [LABEL = "load_label" | LABEL like "label_pattern"];

参数说明

使用示例

• 撤销数据库example_db上，Label为example_db_test_load_label的导入作业。

  CANCEL LOAD
  FROM example_db
  WHERE LABEL = "example_db_test_load_label";

• 撤销数据库example_db上，所有包含example_的导入作业。

  CANCEL LOAD
  FROM example_db
  WHERE LABEL like "example_";

语法

SHOW LOAD
[FROM db_name]
[
   WHERE
   [LABEL [ = "your_label" | LIKE "label_matcher"]]
   [STATE = ["PENDING"|"ETL"|"LOADING"|"FINISHED"|"CANCELLED"|]]
]
[ORDER BY ...]
[LIMIT limit][OFFSET offset];

参数说明

使用示例

• 展示数据库example_db的导入任务，Label中包含字符串2014_01_02，展示保存时间最久的10个。

  SHOW LOAD FROM example_db WHERE LABEL LIKE "2014_01_02" LIMIT 10;

• 展示数据库example_db的导入任务，指定Label为load_example_db_20140102并按LoadStartTime降序排序。

  SHOW LOAD FROM example_db WHERE LABEL = "load_example_db_20140102" ORDER BY LoadStartTime DESC;

• 展示数据库example_db的导入任务，指定Label为load_example_db_20140102，state为loading。

  SHOW LOAD FROM example_db WHERE LABEL = "load_example_db_20140102" AND STATE = "loading";

• 展示数据库example_db的导入任务，按LoadStartTime降序排序，并从偏移量5开始显示10条查询结果。

  SHOW LOAD FROM example_db ORDER BY LoadStartTime DESC limit 5,10;
  SHOW LOAD FROM example_db ORDER BY LoadStartTime DESC limit 10 offset 5;