Serverless Spark DistCp provides efficient distributed data migration and incremental synchronization across storage systems through a SQL interface, with flexible configuration and performance tuning.-E-MapReduce(EMR)-阿里云帮助中心

Serverless Spark provides a distributed data copy feature similar to Hadoop DistCp, letting you efficiently migrate massive volumes of files across storage systems like HDFS, OSS, OSS-HDFS, and Amazon S3. Its pure SQL interface simplifies these operations, allowing you to perform large-scale data copy tasks without writing complex code.

Supported versions

This feature is supported in engine versions esr-3.5.0, esr-4.6.0, and later. We recommend using the latest esr-4.x version.

DISTCP SQL syntax

The DISTCP command recursively copies all files from the source path to the destination path. If the destination directory does not exist, the command creates it automatically. To use this feature, you must set spark.emr.enableDistcp true when you submit a Spark batch job. This parameter cannot be set using the set command in a SQL session, including Kyuubi sessions.

DISTCP FROM 'source_path' TO 'destination_path' [options(key1='v1', key2='v2')];

Parameters

Parameter	Type	Description
`source_path`	STRING	The source path.
`destination_path`	STRING	The destination path.
`OPTIONS(...)`	Key-value pairs	Optional parameters that control copy behavior and filtering rules.

OPTIONS

Parameter

Type

Description

filters

STRING

The path to a text file that contains filtering rules. The path must be an accessible OSS path. Regular expressions follow the Java java.util.regex.Pattern format.

To specify multiple filtering rules, separate them with line breaks.

mode

STRING

Controls the write behavior:

COPY (default): If a file with the same name exists in the destination path, the command deletes the existing file before copying the new one.
UPDATE: Overwrites a destination file only if its size differs from the source file; otherwise, it skips the file.
NOOP: Does not write data to the destination path. This mode is useful for testing the read performance of the source.

Examples

Basic usage: Full copy

DISTCP FROM 'oss://bucket-a/data/input' TO 'oss://bucket-b/data/input-backup';

A successfully executed message indicates that the job is complete.

Incremental synchronization with filtering rules

DISTCP FROM 'oss://bucket-a/data/input' TO 'oss://bucket-b/data/input-backup' OPTIONS(mode='UPDATE', filters='oss://my-bucket/config/exclude_rules.txt');

A successfully executed message indicates that the job is complete.

Previewing execution plans

Use EXPLAIN to preview the execution plan of a DISTCP operation. This lets you estimate resource consumption and verify your configuration.

EXPLAIN DISTCP FROM 'src_path' TO 'dest_path';

The output details the execution plan, including the estimated number of files and total data volume, the actual number of tasks to be generated, and whether filtering rules from the filters file were loaded successfully.

Task concurrency configurations

The following two Spark configurations determine the task concurrency for a DISTCP job:

Parameter	Description
`spark.sql.files.maxPartitionBytes`	The maximum size of a data block that a single task can process. A smaller value increases concurrency, although more tasks can add scheduling overhead.
`spark.sql.files.openCostInBytes`	The estimated cost of opening a file, used to balance small-file merging. A higher value reduces excessive partitioning for small files.

Tuning Recommendations

Large files: Increase maxPartitionBytes to reduce the number of tasks.
Numerous small files: Decrease openCostInBytes to increase concurrency and avoid serial reads.

FAQ

Q1: At which stage are files deleted? If a job terminates unexpectedly, will the files scheduled for deletion in the destination path be lost?
A: Files are deleted during the job commit phase. If a job terminates unexpectedly before the commit, the files scheduled for deletion are retained in the destination path.

Q2: Does DISTCP support copying data to other object storage systems, such as Amazon S3?
A: Yes. DISTCP supports mainstream object storage systems, including Amazon S3, Tencent Cloud COS, Huawei Cloud OBS, Alibaba Cloud OSS, and OSS-HDFS. To access a third-party storage system, you must configure the required authentication parameters and client dependencies when submitting the job. For details, see Cross-cloud adaptation.