ApsaraDB for SelectDB：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;