ApsaraDB for SelectDB:Use Broker Load to import data

Benefits

Broker Load provides the following benefits:

• Large amounts of data: Broker Load can be used to import hundreds of GB of offline data at a time.

• Asynchronously high concurrency: Broker Load can be used to import data in asynchronous mode without data blocking. This improves the cluster resource usage.

• High compatibility: Broker Load can be used to read data from remote storage systems such as HDFS and Amazon S3.

• Ease of use:

  • You can create a Broker Load job over the MySQL protocol to import data.

  • You can execute the SHOW LOAD statement to monitor the data import progress and result in real time.

Scenarios

Broker Load can be used to efficiently read large amounts of data from distributed storage systems such as HDFS, OSS, and Amazon S3.

• The data import timeliness for small amounts of data, such as 100 MB of data, is at the 10-second level.

• The data import timeliness for large amounts of data, such as 100 GB of data, is at the 10-minute level.

Create a Broker Load job

A Broker Load job uses a broker to read and import data from a remote storage system such as HDFS or Amazon S3 into a table of an ApsaraDB for SelectDB instance.

Syntax

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

Parameters

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

[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", ...)]

Examples

1. Create a table into which you want to import data in an ApsaraDB for SelectDB instance. Sample code:

   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. Create a file to import data.

   • Create a file named file1.txt that contains the following content:

     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

   • Create a file named file2.csv that contains the following content:

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

3. Import the data of the files into the table.

   • Import the file1.txt file from HDFS. Sample code:

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

     Import the file1.txt file into the test_table table. Separate columns with commas (,). When you import data from HDFS, you must specify the fs.defaultFS property in the broker_properties parameter to ensure that the system can connect to an HDFS cluster as expected and find the corresponding file.

   • Import the two files into two tables from HDFS. Sample code:

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

     Import the file1.txt file into the test_table table. Import the file2.csv file into the test_table2 table, and increment the values in the age column by 1 based on the values from the temp_age column in the file2.csv file.

   • Import a batch of data from an HDFS cluster that is deployed in high-availability (HA) mode. Sample code:

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

     Use the default delimiter \\x01 of Hive, and use a wildcard (*) to specify all files in the data directory.

   • Filter data in the file1.txt file to import only data rows that meet the filter condition. Sample code:

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

     Only rows whose value of the age column is smaller than 20 are imported.

   • Import the file1.txt file from HDFS. Specify the timeout period and filtering ratio of the import. Import all rows from the file except the rows whose value of the age column is smaller than 20. Sample code:

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

     Import data in MERGE mode. The test_table table must use the Unique Key model. If the value of the age column is less than 20 in a row to be imported, the row is marked for deletion. The timeout period for an import job is 3,600 seconds, and the maximum filtering ratio for error data rows is 10%.

Cancel a Broker Load job

If a Broker Load job is not in the CANCELLED or FINISHED state, you can manually cancel the job. You must specify the label of the import job to be canceled. After an import job is canceled, the data written in the job is rolled back and does not take effect.

Syntax

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

Parameters

Examples

• Cancel an import job whose label is example_db_test_load_label from the example_db database.

  CANCEL LOAD
  FROM example_db
  WHERE LABEL = "example_db_test_load_label";

• Cancel import jobs whose label contains example_ from the example_db database.

  CANCEL LOAD
  FROM example_db
  WHERE LABEL like "example_";

Query the status of a Broker Load job

Broker Load is an asynchronous import method. The successful execution of an import statement indicates only the successful submission of a Broker Load job, rather than the completion of the data import. To query the status of a Broker Load job, execute the SHOW LOAD statement.

Syntax

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

Parameters

Best practices

• Query the status of an import job

  Broker Load is an asynchronous import method. The successful execution of an import statement indicates only the successful submission of a Broker Load job, rather than the completion of the data import. To query the status of an import job, execute the SHOW LOAD statement.

• Cancel an import job

  To cancel an import job that is submitted but is not complete, execute the CANCEL LOAD statement. After an import job is canceled, the data written in the job is rolled back and does not take effect.

• Label, import transaction, and multi-table atomicity

  All import jobs in ApsaraDB for SelectDB are atomic in nature. The atomicity is also ensured when data is imported into multiple tables in an import job. In addition, ApsaraDB for SelectDB uses labels to ensure that the imported data is not lost or duplicated.

• Mapping, deriving, and filtering of columns

  ApsaraDB for SelectDB supports various operations on column conversion and column filtering in an import statement. Most built-in functions and user-defined functions (UDFs) are supported. For more information, see Converting Source Data.

• Filtering of error data rows

  ApsaraDB for SelectDB allows you to skip some data rows in incorrect formats in an import job. The filtering ratio is specified by the max_filter_ratio parameter. The default value is 0, which specifies a zero-tolerance policy that allows no data to be filtered out. The import job fails if one error data row is found. If you want to ignore some error data rows during the import, you can set this parameter to a value between 0 and 1. This way, ApsaraDB for SelectDB automatically skips the rows in incorrect data formats.

  For more information about how to calculate the tolerance ratio, see Converting Source Data.

• Strict mode

  The strict_mode property is used to specify whether an import job runs in strict mode. If an import job runs in strict mode, the mapping, conversion, and filtering of columns are affected.

• Timeout period

  The default timeout period for a Broker Load job is 4 hours. The timing starts from the time when the import job is submitted. If the import job is not complete within the timeout period, the import job fails.

• Limits on the amount of data and the number of jobs

  We recommend that you import data of less than 100 GB at a time in a Broker Load job. Theoretically, the amount of data that can be imported in an import job is not limited. However, if the amount of data to be imported is excessively large, the import job may run for an extended period of time, and the cost of retry is high in case that the import job fails.

  In addition, the number of nodes in a cluster is limited. Therefore, a limit is set on the maximum amount of data to be imported, which is the number of nodes multiplied by 3 GB. This ensures that system resources are properly used. If you need to import a large amount of data, we recommend that you submit multiple import jobs.

  ApsaraDB for SelectDB limits the number of concurrent import jobs in a cluster, which is in the range of 3 to 10. If the number of import jobs that you submitted is greater than the limit, the excess import jobs wait in a queue. The maximum length of the queue is 100. If more than 100 import jobs are waiting, the excess import jobs are directly rejected.

  Note

  The waiting time also counts in the total period of time for an import job. If a timeout error occurs, the job is canceled. We recommend that you monitor the status of import jobs to properly control the job submission frequency.