Exporting Data from HDFS to OBS

Updated on 2024-11-29 GMT+08:00

View PDF

You can export data in specified tables or partitions in text format through the Broker process or S3 protocol, to remote storage, such as HDFS and object storage.

NOTE:

You are not advised to export a large amount of data at a time. It is recommended that a maximum of dozens of GB data be exported with an export job. Large exports result in many junk files and high retry costs.
If a table contains a large amount of data, you are advised to export the data by partition.
If the FE is restarted or an active/standby switchover occurs when the export job is running, the job will fail and you will need to submit it again.
If an export job fails, the temporary directory __doris_export_tmp_xxx generated in the remote storage and generated files will not be deleted. You need to manually delete them.
If an export job is successfully executed, the __doris_export_tmp_xxx directory generated in the remote storage may be retained or cleared based on the file system semantics of the remote storage.
For example, in object storage (using S3 protocol), after the last file in a directory is moved by the rename operation, the directory will also be deleted. If the directory is not cleared by the system, you can clear it.
After the export ends (successful or failed), if the FE is restarted or an active/standby switchover occurs, some job information displayed in SHOW EXPORT will be lost and cannot be viewed.
This function exports only Base table data and does not export Rollup Index data.
Export jobs scan data and occupy I/O resources, which may cause query latency.

Syntax

Exporting Doris data to HDFS

Kerberos authentication is enabled for the cluster (the cluster is in security mode)
EXPORT TABLE db1.tbl1

PARTITION (p1,p2)

[WHERE [expr]]

TO "hdfs://IP address of the active NameNode instance:RPC port number/tmp/export/"

PROPERTIES

(

"label" = "mylabel",

"column_separator"=",",

"columns" = "col1,col2",

"exec_mem_limit"="2147483648",

"timeout" = "3600"

)

WITH BROKER "broker_name"

(

"hadoop.security.authentication"="kerberos",

"kerberos_principal"="doris/hadoop.hadoop.com@HADOOP.COM",

"kerberos_keytab"="${BIGDATA_HOME}/FusionInsight_Doris_*/install/FusionInsight-Doris-*/doris-fe/bin/doris.keytab"

);
Kerberos authentication is disabled for the cluster (the cluster is in normal mode)
EXPORT TABLE db1.tbl1

PARTITION (p1,p2)

[WHERE [expr]]

TO "hdfs://IP address of the active NameNode instance:RPC port number/tmp/export/"

PROPERTIES

(

"label" = "mylabel",

"column_separator"=",",

"columns" = "col1,col2",

"exec_mem_limit"="2147483648",

"timeout" = "3600"

)

WITH BROKER "broker_name"

(

"username" = "user",

"password" = "passwd"

);

To view the IP address of the active NameNode instance, log in to FusionInsight Manager and choose Cluster > Services > HDFS > Instances. Commands carrying authentication passwords pose security risks. Disable historical command recording before running such commands to prevent information leakage.

You can log in to FusionInsight Manager, choose Cluster > Services > HDFS > Configurations, and search for dfs.namenode.rpc.port to view the RPC port.

Table 1 describes other parameters.

**Table 1** Parameters in the command for exporting Doris data to HDFS
Parameter	Description
label	Label of the export job. You can use this identifier to view the job status.
column_separator	Column separator. The default value is \t. Invisible characters are supported, for example, '\x07'.
columns	Columns to be exported. Use commas (,) to separate them. If this parameter is not set, all columns in the table will be exported by default.
line_delimiter	Line delimiter. The default value is \n. Invisible characters are supported, for example, '\x07'.
exec_mem_limit	Maximum memory can be used by a query plan in an export job on a BE. The default value is 2 GB, in bytes.
timeout	Job timeout period. The default value is 2 hours. The unit is second.
tablet_num_per_task	Maximum number of shards allocated to each query plan. The default value is 5.

Viewing the status of an export job.

After submitting a job, you can run the SHOW EXPORT; command to query the status of the export job. The following is an example:

JobId: 14008
State: FINISHED
Progress: 100%
TaskInfo: {"partitions":["*"],"exec mem limit":2147483648,"column separator":",","line delimiter":"\n","tablet num":1,"broker":"hdfs","coord num":1,"db":"default_cluster:db1","tbl":"tbl3"}
Path: hdfs://host/path/to/export/
CreateTime: 2019-06-25 17:08:24
StartTime: 2019-06-25 17:08:28
FinishTime: 2019-06-25 17:08:34
Timeout: 3600
ErrorMsg: NULL
1 row in set (0.01 sec)

**Table 2** Export job information
Parameter	Description
JobId	Job ID, which is unique.
State	Job status. PENDING: The job is to be scheduled. EXPORTING: Data is being exported. FINISHED: The export job is successfully executed. ANCELLED: The export job fails to be executed.
Progress	Work progress based on the query plan. Assuming there are 10 threads in total and 3 have been completed, the progress will be 30%.
TaskInfo	Job information displayed in JSON format db: database name tbl: table name partitions: partitions to be exported. * indicates all partitions. exec mem limit: memory usage limit for the query plan, in bytes. column separator: column separator for the exported file line delimiter: line delimiter for the exported file tablet num: total number of tablets broker: name of the Broker coord num: number of query plans
Path	Export path on the remote storage
CreateTime/StartTime/FinishTime	Creation time, scheduling start time, and end time of a job
Timeout	Job timeout period, in seconds. The time starts from CreateTime.
ErrorMsg	If an error occurs, ErrorMsg displays the cause of the error.

Canceling an export task
After submitting a job, you can run the CANCEL_EXPORT command to cancel the export job. The cancellation command is as follows:

CANCEL EXPORT

FROM example_db

WHERE LABEL like "%example%";

Prerequisite

A cluster containing the Doris service has been created, and all services in the cluster are running properly.
The nodes to be connected to the Doris database can communicate with the MRS cluster.
A user with Doris management permission has been created.
- Kerberos authentication is enabled for the cluster (the cluster is in security mode)
  Log in to FusionInsight Manager, create a human-machine user, for example, dorisuser, create a role with Doris administrator permissions, and bind the role to the user.
  
  Log in to FusionInsight Manager as the new user dorisuser and change the initial password.
- Kerberos authentication is disabled for the cluster (the cluster is in normal mode)
  After connecting to Doris as user admin, create a role with administrator permissions, and bind the role to the user.
The MySQL client has been installed. For details, see Installing a MySQL Client.

Export Job Example

Log in to the node where MySQL is installed and run the following command to connect to the Doris database:

If Kerberos authentication is enabled for the cluster (the cluster is in security mode), run the following command to connect to the Doris database:

export LIBMYSQL_ENABLE_CLEARTEXT_PLUGIN=1

mysql -uDatabase login username -pDatabase login password -PConnection port for FE queries -hIP address of the Doris FE instance
NOTE:
- To obtain the query connection port of the Doris FE instance, you can log in to FusionInsight Manager, choose Cluster > Services > Doris > Configurations, and query the value of query_port of the Doris service.
- To obtain the IP address of the Doris FE instance, log in to FusionInsight Manager of the MRS cluster and choose Cluster > Services > Doris > Instances to view the IP address of any FE instance.
- You can also use the MySQL connection software or Doris web UI to connect to the database.
Run the following statement to create a database:

create database if not exists example_db;
Run the following statement to create a table:

CREATE TABLE example_db.test_export_tbl (

`c1` int NOT NULL,

`c2` int NOT NULL,

`c3` string NOT NULL,

`c4` date NOT NULL

) ENGINE=OLAP

DUPLICATE KEY(`c1`, `c2`)

DISTRIBUTED BY HASH(`c1`) BUCKETS 1;
Run the following statement to insert data:

insert into example_db.test_export_tbl values(1,1,1,"2020-02-21"),(2,2,2,"2020-03-21"),(3,3,3,"2020-04-21");
Run the following statement to export data from the test_export_tbl table to HDFS:

EXPORT TABLE example_db.test_export_tbl

TO "hdfs://IP address of the active NameNode instance:RPC port number/tmp/export/"

PROPERTIES

(

"label" = "label_exporthdfs_20230218031",

"column_separator"=",",

"columns" = "c1,c2,c3,c4",

"exec_mem_limit"="2147483648",

"timeout" = "3600"

)

with broker "broker_192_168_67_78"

(

"hadoop.security.authentication"="kerberos",

"kerberos_principal"="doris/hadoop.hadoop.com@HADOOP.COM",

"kerberos_keytab"="${BIGDATA_HOME}/FusionInsight_Doris_8.3.1/install/FusionInsight-Doris-2.0.3/doris-fe/bin/doris.keytab"

);
Run the following statement to query the status of the export job:

SHOW EXPORT;

Related Configuration Parameters

Log in to FusionInsight Manager, choose Cluster > Services > Doris > Configurations > FE(Role) > Customization, and add the following parameters to fe.conf.customized.configs:

export_checker_interval_second: scheduling interval of the export job scheduler. The default value is 5 seconds. After setting this parameter, restart the FE.
export_running_job_num_limit: maximum number of running export jobs Exceeding jobs wait and are in the PENDING state. The default value is 5, which can be adjusted at runtime.
export_task_default_timeout_second: default timeout period of an export job. The default value is 2 hours, which can be adjusted at runtime.
export_tablet_num_per_task: maximum number of shards that a query plan is responsible for. The default value is 5.
label: user-customized label of an export task. If this parameter is not specified, a label will be automatically generated.