Use Jindo DistCp - E-MapReduce - Alibaba Cloud Documentation Center

Jindo DistCp is a distributed data copy tool built on MapReduce. Use it to copy files between Hadoop Distributed File System (HDFS) and Object Storage Service (OSS), between OSS buckets, or from Amazon S3 to OSS — with options for filtering, compression, incremental copy, and performance tuning.

Prerequisites

Before you begin, ensure that you have:

Java Development Kit (JDK) 8 installed on the machine where you run commands
An E-MapReduce (EMR) cluster of version 3.28.0 or later (Create a cluster)

Parameters

Run jindo distcp --help on the EMR master node to see all available parameters.

The table below lists each parameter, whether it is required, and what it does.

Parameter	Required	Description
`--src`	Yes	Source directory to copy files from
`--dest`	Yes	Destination directory to copy files to
`--parallelism`	No	Number of parallel reduce tasks. Maps to `mapreduce.job.reduces`. Default: `7`
`--srcPattern`	No	Regular expression to filter source files. Must match the full path
`--deleteOnSuccess`	No	Delete source files after a successful copy
`--outputCodec`	No	Compression codec to apply to copied files. Values: `gzip`, `gz`, `lzo`, `lzop`, `snappy`, `none`, `keep`. Default: `keep`
`--outputManifest`	No	Name of the manifest file to generate. Lists all copied files (destination path, source path, file size). Only gzip format is supported
`--requirePreviousManifest`	No	Set to `false` to generate a manifest file without requiring a previous one
`--previousManifest`	No	Path to an existing manifest file listing previously copied files
`--copyFromManifest`	No	Copy files listed in a manifest file instead of scanning a directory
`--srcPrefixesFile`	No	Path to a file containing source URI prefixes, one per line. Copies files from multiple directories in one job
`--groupBy`	No	Regular expression pattern to group input files for merging
`--targetSize`	No	Target size for merged output files, in MB
`--enableBalancePlan`	No	Optimize task allocation when file sizes are similar. Cannot be combined with `--groupBy` or `--targetSize`
`--enableDynamicPlan`	No	Optimize task allocation when most files are small and file sizes vary significantly. Cannot be combined with `--groupBy` or `--targetSize`
`--enableTransaction`	No	Enable job-level transaction support to ensure integrity
`--diff`	No	Compare source and destination file lists and report differences
`--ossKey`	No	AccessKey ID for OSS access
`--ossSecret`	No	AccessKey Secret for OSS access
`--ossEndPoint`	No	OSS endpoint
`--policy`	No	OSS storage policy for written data. Values: `archive`, `ia` (Infrequent Access)
`--cleanUpPending`	No	Clean up incomplete multipart uploads (identified by upload ID) after the job completes
`--queue`	No	YARN queue name for the copy job
`--bandwidth`	No	Bandwidth limit per map/reduce task, in MB/s
`--s3Key`	No	Amazon S3 access key
`--s3Secret`	No	Amazon S3 secret key
`--s3EndPoint`	No	Amazon S3 endpoint

Limitations

--enableBalancePlan and --enableDynamicPlan cannot be combined with --groupBy or --targetSize.
--outputManifest only supports gzip format for the manifest file.
--diff does not report accurate file size differences when compression or decompression was applied during the copy.
When the destination is an HDFS directory, --dest must use one of these formats: /path, hdfs://hostname:port/path, or hdfs://headerIp:port/path. The formats hdfs:///path and hdfs:/path are not supported.
To use the LZO codec in an open-source Hadoop cluster, install the native gplcompression library and the hadoop-lzo package.

Copy files from HDFS to OSS

Log in to the master node of your EMR cluster via SSH (Connect to the master node in SSH mode), then run commands from there.

The basic copy command copies all files in the source directory to the destination:

jindo distcp --src /opt/tmp --dest oss://yang-hhht/tmp

If the destination directory does not exist, Jindo DistCp creates it automatically.

To increase copy throughput, set --parallelism to a value higher than the default of 7:

jindo distcp --src /opt/tmp --dest oss://yang-hhht/tmp --parallelism 20

Filter files by pattern

Use --srcPattern with a regular expression to copy only matching files. The pattern must match the full file path.

For example, to copy only .log files from /data/incoming/hourly_table:

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --srcPattern '.*\.log' --parallelism 20

To verify, list the source directory first:

hdfs dfs -ls /data/incoming/hourly_table/2017-02-01/03

Output:

Found 6 items
-rw-r-----   2 root hadoop       2252 2020-04-17 20:42 /data/incoming/hourly_table/2017-02-01/03/000151.sst
-rw-r-----   2 root hadoop       4891 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/1.log
-rw-r-----   2 root hadoop       4891 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/2.log
-rw-r-----   2 root hadoop       4891 2020-04-17 20:42 /data/incoming/hourly_table/2017-02-01/03/OPTIONS-000109
-rw-r-----   2 root hadoop       1016 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/emp01.txt
-rw-r-----   2 root hadoop       1016 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/emp06.txt

After the copy, only the two .log files appear in the destination:

hdfs dfs -ls oss://yang-hhht/hourly_table/2017-02-01/03

Output:

Found 2 items
-rw-rw-rw-   1       4891 2020-04-17 20:52 oss://yang-hhht/hourly_table/2017-02-01/03/1.log
-rw-rw-rw-   1       4891 2020-04-17 20:52 oss://yang-hhht/hourly_table/2017-02-01/03/2.log

Delete source files after copying

Add --deleteOnSuccess to remove source files once they are successfully copied to the destination:

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --deleteOnSuccess --parallelism 20

Compress files during copy

Use --outputCodec to compress files on the fly as they are copied. Supported values: gzip, gz, lzo, lzop, snappy, none, keep (default).

keep: Copy files without changing compression.
none: Copy files uncompressed. If files are already compressed, they are decompressed.

Example using gzip compression:

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --outputCodec=gz --parallelism 20

List the destination to confirm all files are compressed:

hdfs dfs -ls oss://yang-hhht/hourly_table/2017-02-01/03

Output:

Found 6 items
-rw-rw-rw-   1        938 2020-04-17 20:58 oss://yang-hhht/hourly_table/2017-02-01/03/000151.sst.gz
-rw-rw-rw-   1       1956 2020-04-17 20:58 oss://yang-hhht/hourly_table/2017-02-01/03/1.log.gz
-rw-rw-rw-   1       1956 2020-04-17 20:58 oss://yang-hhht/hourly_table/2017-02-01/03/2.log.gz
-rw-rw-rw-   1       1956 2020-04-17 20:58 oss://yang-hhht/hourly_table/2017-02-01/03/OPTIONS-000109.gz
-rw-rw-rw-   1        506 2020-04-17 20:58 oss://yang-hhht/hourly_table/2017-02-01/03/emp01.txt.gz
-rw-rw-rw-   1        506 2020-04-17 20:58 oss://yang-hhht/hourly_table/2017-02-01/03/emp06.txt.gz

Run incremental copies with manifest files

A manifest file records every file copied by a job — including destination path, source path, and file size. Use manifests to track copy history and run incremental jobs that copy only new files.

Generate a manifest for the initial copy

Set --requirePreviousManifest=false to create a manifest without requiring one from a previous job. The manifest is saved in gzip format.

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --outputManifest=manifest-2020-04-17.gz --requirePreviousManifest=false --parallelism 20

To inspect the manifest:

hadoop fs -text oss://yang-hhht/hourly_table/manifest-2020-04-17.gz > before.lst
cat before.lst

Output:

{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/000151.sst","baseName":"2017-02-01/03/000151.sst","srcDir":"oss://yang-hhht/hourly_table","size":2252}
{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/1.log","baseName":"2017-02-01/03/1.log","srcDir":"oss://yang-hhht/hourly_table","size":4891}
{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/2.log","baseName":"2017-02-01/03/2.log","srcDir":"oss://yang-hhht/hourly_table","size":4891}
{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/OPTIONS-000109","baseName":"2017-02-01/03/OPTIONS-000109","srcDir":"oss://yang-hhht/hourly_table","size":4891}
{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/emp01.txt","baseName":"2017-02-01/03/emp01.txt","srcDir":"oss://yang-hhht/hourly_table","size":1016}
{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/emp06.txt","baseName":"2017-02-01/03/emp06.txt","srcDir":"oss://yang-hhht/hourly_table","size":1016}

Copy only new files using a previous manifest

Pass the previous manifest with --previousManifest and generate a new one with --outputManifest. Jindo DistCp skips files already listed in the previous manifest and copies only new files. The new manifest contains the full copy history.

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --outputManifest=manifest-2020-04-18.gz --previousManifest=oss://yang-hhht/hourly_table/manifest-2020-04-17.gz --parallelism 20

To see what was added in this job, diff the two manifests:

hadoop fs -text oss://yang-hhht/hourly_table/manifest-2020-04-18.gz > current.lst
diff before.lst current.lst

Output — two new files were copied:

3a4,5
> {"path":"oss://yang-hhht/hourly_table/2017-02-01/03/5.log","baseName":"2017-02-01/03/5.log","srcDir":"oss://yang-hhht/hourly_table","size":4891}
> {"path":"oss://yang-hhht/hourly_table/2017-02-01/03/6.log","baseName":"2017-02-01/03/6.log","srcDir":"oss://yang-hhht/hourly_table","size":4891}

Re-copy files from a manifest

Use --copyFromManifest with --previousManifest to copy exactly the files listed in a manifest:

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --previousManifest=oss://yang-hhht/hourly_table/manifest-2020-04-17.gz --copyFromManifest --parallelism 20

Copy files from multiple source directories

Use --srcPrefixesFile to copy files from multiple directories in a single job. Create a plain text file listing one source URI prefix per line, then pass it to the command.

For example, given a folders.txt file with this content:

hdfs://emr-header-1.cluster-50466:9000/data/incoming/hourly_table/2017-02-01
hdfs://emr-header-1.cluster-50466:9000/data/incoming/hourly_table/2017-02-02

Run:

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --srcPrefixesFile file:///opt/folders.txt --parallelism 20

Merge small files

Reading large numbers of small files from HDFS is slow. Use --groupBy and --targetSize together to merge small files into larger output files during the copy.

--groupBy: A regular expression that groups input files. Files matching the same group are merged into one output file.
--targetSize: Maximum size of each merged output file, in MB.

For example, to merge all .txt files in hourly_table into files of up to 10 MB each:

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --targetSize=10 --groupBy='.*/([a-z]+).*.txt' --parallelism 20

Source directory (8 files):

Found 8 items
-rw-r-----   2 root hadoop       2252 2020-04-17 20:42 /data/incoming/hourly_table/2017-02-01/03/000151.sst
-rw-r-----   2 root hadoop       4891 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/1.log
-rw-r-----   2 root hadoop       4891 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/2.log
-rw-r-----   2 root hadoop       4891 2020-04-17 21:08 /data/incoming/hourly_table/2017-02-01/03/5.log
-rw-r-----   2 root hadoop       4891 2020-04-17 21:08 /data/incoming/hourly_table/2017-02-01/03/6.log
-rw-r-----   2 root hadoop       4891 2020-04-17 20:42 /data/incoming/hourly_table/2017-02-01/03/OPTIONS-000109
-rw-r-----   2 root hadoop       1016 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/emp01.txt
-rw-r-----   2 root hadoop       1016 2020-04-17 20:47 /data/incoming/hourly_table/2017-02-01/03/emp06.txt

Destination directory — the two .txt files are merged into one:

Found 1 items
-rw-rw-rw-   1       2032 2020-04-17 21:18 oss://yang-hhht/hourly_table/2017-02-01/03/emp2

Tune copy performance

Balance task allocation for mixed file sizes

If the source contains both small and large files and the size difference within each group is not significant, use --enableBalancePlan to distribute tasks more evenly across reducers:

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --enableBalancePlan --parallelism 20

Cannot be combined with --groupBy or --targetSize.

Optimize for mostly small files with high size variance

If most files are small and sizes vary significantly, use --enableDynamicPlan to assign tasks dynamically:

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --enableDynamicPlan --parallelism 20

Cannot be combined with --groupBy or --targetSize.

Limit bandwidth usage

Use --bandwidth to prevent a copy job from consuming excessive network bandwidth. The value is in MB/s and applies per map/reduce task:

jindo distcp --src /data/incoming/hourly_table --dest oss://<your_bucket>/hourly_table --bandwidth 100

Assign to a specific YARN queue

Use --queue to route the job to a particular YARN queue:

jindo distcp --src /data/incoming/hourly_table --dest oss://<your_bucket>/hourly_table --queue yarnqueue

Enable transaction support

Use --enableTransaction to ensure job-level integrity and transaction support among jobs:

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --enableTransaction --parallelism 20

Verify copy completeness

After a copy job, use --diff to compare the source and destination file lists:

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --diff

If all files were copied successfully:

INFO distcp.JindoDistCp: distcp has been done completely

If some files were not copied, Jindo DistCp generates a manifest file in the destination directory listing the missing files. Use --copyFromManifest and --previousManifest to copy the remaining files:

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --previousManifest=file:///opt/manifest-2020-04-17.gz --copyFromManifest --parallelism 20

--diff does not report accurate file size differences if compression or decompression was applied during the copy.

Access OSS with an AccessKey pair

By default, EMR clusters access OSS without requiring explicit credentials. If you run Jindo DistCp outside an EMR cluster, or if AccessKey-free access is not supported, specify your credentials with --ossKey, --ossSecret, and --ossEndPoint:

jindo distcp --src /data/incoming/hourly_table --dest oss://<your_bucket>/hourly_table --ossKey <your-access-key-id> --ossSecret <your-access-key-secret> --ossEndPoint oss-cn-hangzhou.aliyuncs.com --parallelism 20

Write to OSS Archive or Infrequent Access storage

Use --policy to write data directly to a lower-cost OSS storage class during the copy:

archive: Archive storage class

jindo distcp --src /data/incoming/hourly_table --dest oss://<your_bucket>/hourly_table --policy archive --parallelism 20

ia: Infrequent Access (IA) storage class

jindo distcp --src /data/incoming/hourly_table --dest oss://<your_bucket>/hourly_table --policy ia --parallelism 20

Clean up incomplete uploads

When a copy job is interrupted, files that were partially uploaded may remain in the destination bucket. OSS tracks these by upload ID, and they are not visible in normal directory listings.

Add --cleanUpPending to automatically remove these incomplete uploads when the job finishes:

jindo distcp --src /data/incoming/hourly_table --dest oss://<your_bucket>/hourly_table --cleanUpPending --parallelism 20

Alternatively, clean them up manually in the OSS console.

Copy from Amazon S3 to OSS

Specify the S3 credentials and endpoint with --s3Key, --s3Secret, and --s3EndPoint:

jindo distcp jindo-distcp-2.7.3.jar --src s3a://yourbucket/ --dest oss://<your_bucket>/hourly_table --s3Key <your-s3-key> --s3Secret <your-s3-secret> --s3EndPoint s3-us-west-1.amazonaws.com

To avoid passing credentials on every command, configure them in the Hadoop core-site.xml file:

<configuration>
    <property>
        <name>fs.s3a.access.key</name>
        <value>xxx</value>
    </property>
    <property>
        <name>fs.s3a.secret.key</name>
        <value>xxx</value>
    </property>
    <property>
        <name>fs.s3.endpoint</name>
        <value>s3-us-west-1.amazonaws.com</value>
    </property>
</configuration>

Then run the command without inline credentials:

jindo distcp /tmp/jindo-distcp-2.7.3.jar --src s3://smartdata1/ --dest s3://smartdata1/tmp --s3EndPoint s3-us-west-1.amazonaws.com

Check copy counters

After a job completes, check the DistCp Counters in the MapReduce job output to verify how much data was transferred:

Distcp Counters
        Bytes Destination Copied=11010048000
        Bytes Source Read=11010048000
        Files Copied=1001

Shuffle Errors
        BAD_ID=0
        CONNECTION=0
        IO_ERROR=0
        WRONG_LENGTH=0
        WRONG_MAP=0
        WRONG_REDUCE=0

If compression or decompression was applied during the copy, Bytes Destination Copied and Bytes Source Read may differ.