E-MapReduce:Use Jindo DistCp

Basic copy

Copy all files from /opt/tmp in HDFS to an OSS bucket:

jindo distcp --src /opt/tmp --dest oss://<yourBucketName>/tmp

Replace <yourBucketName> with the name of your OSS bucket. If you do not specify a root directory in --dest, Jindo DistCp creates one automatically.

Adjust parallelism

--parallelism controls how many reduce tasks run in parallel. The default is 7. Increase it when you have more cluster resources available:

jindo distcp --src /opt/tmp --dest oss://<yourBucketName>/tmp --parallelism 20

Filter files by pattern

--srcPattern accepts a regular expression that must match the full file path. To copy only .log files from /data/incoming/hourly_table:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --srcPattern '.*\.log' --parallelism 20

Verify the result:

hdfs dfs -ls oss://<yourBucketName>/hourly_table/2017-02-01/03

Expected output — only .log files are copied:

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

Delete source files after copy

--deleteOnSuccess removes source files once the copy succeeds:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --deleteOnSuccess --parallelism 20

Compress files during copy

--outputCodec compresses files as they are copied. Valid values: gzip, gz, lzo, lzop, snappy, none, keep (default).

• none: decompress files that are already compressed

• keep: copy files without changing their compression

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --outputCodec=gz --parallelism 20

Verify that files in the destination are compressed:

hdfs dfs -ls oss://<yourBucketName>/hourly_table/2017-02-01/03

Expected output:

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

To use the LZO codec in an open source Hadoop cluster, install the gplcompression native library and the hadoop-lzo package.

Copy files in multiple folders

--srcPrefixesFile points to a file where each line is a source URI prefix. Jindo DistCp copies all files under the listed prefixes:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --srcPrefixesFile file:///opt/folders.txt --parallelism 20

Sample folders.txt:

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

Merge small files

Reading large numbers of small files from HDFS degrades query performance. Use --groupBy and --targetSize to merge them into larger files.

--groupBy is a regular expression that groups files by a captured pattern. --targetSize sets the maximum output file size in MB.

The following example merges all .txt files (grouped by name prefix) into files no larger than 10 MB:

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

Verify the result:

hdfs dfs -ls oss://<yourBucketName>/hourly_table/2017-02-01/03/

Expected output — two .txt files merged into one:

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

--groupBy and --targetSize cannot be used with --enableBalancePlan or --enableDynamicPlan.

Optimize task allocation

Use one of these options to improve copy performance based on your file size distribution. Neither option can be used with --groupBy or --targetSize.

• --enableBalancePlan: Use when both small and large files are to be copied but the file sizes are not significantly different among small files and among large files. Distributes tasks evenly across reducers.

  jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --enableBalancePlan --parallelism 20

• --enableDynamicPlan: Use when file sizes vary significantly and most files are small. Allocates tasks dynamically to avoid reducer imbalance.

  jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --enableDynamicPlan --parallelism 20

Enable transaction support

--enableTransaction ensures job-level atomicity:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --enableTransaction --parallelism 20

Exclude files with filters

--filters points to a file where each line is a regular expression. Files with paths matching any expression are excluded from both copy and --diff operations.

hadoop jar jindo-distcp-3.5.0.jar --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --filters /path/to/filterfile.txt --parallelism 20

Sample filter file:

.*\.tmp.
.*\.staging.*

Files whose paths contain .tmp or .staging are skipped.

Limit bandwidth

--bandwidth caps the bandwidth used by each node (in MB/s), preventing the task from consuming all available network capacity:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --bandwidth 100 --parallelism 20

Specify a YARN queue

--queue assigns the DistCp job to a specific YARN queue:

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

Generate a manifest

Set --requirePreviousManifest=false to create the first manifest. The file is always compressed in GZ format:

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

Read the manifest:

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

Each line is a JSON object describing one file:

{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/000151.sst","baseName":"2017-02-01/03/000151.sst","srcDir":"oss://<yourBucketName>/hourly_table","size":2252}
{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/1.log","baseName":"2017-02-01/03/1.log","srcDir":"oss://<yourBucketName>/hourly_table","size":4891}
...

Copy newly added files

Pass the previous manifest to --previousManifest so that Jindo DistCp only copies files not already listed:

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

Compare manifests to see what was newly copied:

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

Output shows the two newly added files:

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

Replay a manifest

To copy exactly the files listed in a previous manifest:

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

Check for differences

--diff compares source and destination file lists. Run it after a copy to verify completeness:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --diff

If all files are copied:

INFO distcp.JindoDistCp: Jindo DistCp job exit with 0

If some files are missing, Jindo DistCp generates a manifest in the destination directory listing the missing files. Use --copyFromManifest and --previousManifest to copy them:

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

--diff does not return accurate file size differences if files were compressed or decompressed during the copy. For HDFS destinations, use /path, hdfs://hostname:ip/path, or hdfs://headerIp:ip/path. Formats like hdfs:///path and hdfs:/path are not supported.

Sync incrementally

--update skips the files and directories that are the same as those in the destination and copies only new or changed files and directories:

hadoop jar jindo-distcp-3.5.0.jar --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --update --parallelism 20