E-MapReduce:JindoDistCp user guide

What is JindoDistCp

JindoDistCp is a distributed file copy tool developed by the Alibaba Cloud Data Lake Storage team for large-scale data transfers within and between clusters. It uses MapReduce to distribute files, handle errors, and recover from failures. It takes a list of files and directories as input for MapReduce tasks, where each task copies a portion of the source list. It fully supports data copy scenarios between the Hadoop Distributed File System (HDFS), OSS-HDFS, OSS, and S3. It provides various custom copy parameters and strategies. It is optimized for copying data from HDFS to OSS-HDFS. Using a custom CopyCommitter, it performs No-Rename copies and ensures data consistency upon completion. Its features are fully aligned with those of S3 DistCp and HDFS DistCp. It offers significant performance improvements over HDFS DistCp. JindoDistCp is designed to be an efficient, stable, and secure data copy tool.

Environment requirements

• JDK 1.8.0 or later.

• Hadoop 2.3 or later. You must download the latest version of the `jindo-distcp-tool-x.x.x.jar` file. This JAR file is included in the `jindosdk-${version}.tar.gz` package. After you decompress the package, you can find the JAR file in the `tools/` directory. For more information, see JindoData downloads.

  Note

  JindoDistCp is deployed on clusters that run EMR V5.6.0 or later and EMR V3.40.0 or later. You can find the `jindo-distcp-tool-x.x.x.jar` file in the `/opt/apps/JINDOSDK/jindosdk-current/tools` directory.

Parameters

--src and --dest (required)

• `--src`: Specifies the path of the source files.

• `--dest`: Specifies the path of the destination files.

hadoop jar jindo-distcp-tool-${version}.jar --src /data/hourly_table  --dest oss://example-oss-bucket/hourly_table

You can specify the destination directory using the `dest` path. For example, the preceding command copies files from `/data/hourly_table` to the `hourly_table` directory in the `example-oss-bucket` bucket. This behavior is different from that of Hadoop DistCp. By default, JindoDistCp copies all files from the source directory to the specified destination path but does not include the root directory of the source. You can specify a root directory in the destination path. If the directory does not exist, it is automatically created.

hadoop jar jindo-distcp-tool-${version}.jar --src /test.txt --dest oss://example-oss-bucket/tmp

Use --bandWidth

`--bandWidth`: Specifies the bandwidth that a single node can use for the DistCp task, in MB. This parameter prevents a single node from consuming too much bandwidth.

jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --bandWidth 6

Use --codec

Source files are often stored in OSS or OSS-HDFS as uncompressed text, which is not ideal for storage costs or data analysis. You can use the `--codec` option to efficiently store data by compressing files online.

jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --codec gz

[root@emr-header-1 opt]# hdfs dfs -ls oss://example-oss-bucket/hourly_table/2017-02-01/03
Found 6 items
-rw-rw-rw-   1        938 2020-04-17 20:58 oss://example-oss-bucket/hourly_table/2017-02-01/03/000151.sst.gz
-rw-rw-rw-   1       1956 2020-04-17 20:58 oss://example-oss-bucket/hourly_table/2017-02-01/03/1.log.gz
-rw-rw-rw-   1       1956 2020-04-17 20:58 oss://example-oss-bucket/hourly_table/2017-02-01/03/2.log.gz
-rw-rw-rw-   1       1956 2020-04-17 20:58 oss://example-oss-bucket/hourly_table/2017-02-01/03/OPTIONS-000109.gz
-rw-rw-rw-   1        506 2020-04-17 20:58 oss://example-oss-bucket/hourly_table/2017-02-01/03/emp01.txt.gz
-rw-rw-rw-   1        506 2020-04-17 20:58 oss://example-oss-bucket/hourly_table/2017-02-01/03/emp06.txt.gz

Use --filters

`--filters`: Specifies a file that contains filter rules.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --filters filter.txt

For example, if the `filter.txt` file contains .*test.*, files whose paths contain the string "test" are not copied to OSS.

Use --srcPrefixesFile

`--srcPrefixesFile`: Specifies a file that contains inclusion rules.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --srcPrefixesFile prefixes.txt

For example, if the `prefixes.txt` file contains .*test.*, only files whose paths contain the string "test" are copied to OSS.

Use --parallelism

`--parallelism`: Specifies the `mapreduce.job.maps` parameter for the MapReduce task. The default value of this parameter in an EMR environment is 10. You can customize the value of this parameter based on your cluster resources to control the concurrency of the DistCp task.

 jindo-distcp-tool-${version}.jar --src /opt/tmp --dest oss://example-oss-bucket/tmp --parallelism 20

Use --taskBatch

`--taskBatch`: Specifies the number of files to be processed by each DistCp task. The default value is 1.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --taskBatch 1

Use --tmp

`--tmp`: Specifies a temporary directory in HDFS to store temporary data. The default value is `/tmp`, which corresponds to `hdfs:///tmp/`.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table  --tmp /tmp

Configure an AccessKey to access OSS or OSS-HDFS

`--hadoopConf`: If you are not in an EMR environment or if the passwordless access service has issues, you can use this option to specify an AccessKey to access OSS or OSS-HDFS.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --hadoopConf fs.oss.accessKeyId=yourkey --hadoopConf fs.oss.accessKeySecret=yoursecret

<configuration>
    <property>
        <name>fs.oss.accessKeyId</name>
        <value>xxx</value>
    </property>

    <property>
        <name>fs.oss.accessKeySecret</name>
        <value>xxx</value>
    </property>
</configuration>

Use --disableChecksum

`--disableChecksum`: Disables file checksum verification.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --disableChecksum

Use --deleteOnSuccess

`--deleteOnSuccess`: Moves data instead of copying it. This option is similar to an `mv` operation. It first copies the files and then deletes them from the source.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --deleteOnSuccess

Use --enableTransaction

`--enableTransaction`: By default, JindoDistCp ensures task-level integrity. You can use this parameter to ensure job-level integrity and enable transactional support between jobs.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --enableTransaction

Use --ignore

`--ignore`: Ignores exceptions that occur during data migration. Errors do not interrupt the task. Instead, the errors are reported as JindoCounter values. If CMS is enabled, a notification is also sent.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --ignore

Use --diff

`--diff`: Compares the files in the source and destination. If a source file is not synchronized to the destination, a file that contains the differences is generated in the current directory. If your JindoDistCp task involves compression or decompression, `--diff` cannot show the correct file differences because the file size is changed during the process.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --diff

JindoCounter
DIFF_FILES=1

If your `--dest` is an HDFS path, the `/path`, `hdfs://hostname:ip/path`, and `hdfs://headerIp:ip/path` formats are supported. The `hdfs:///path`, `hdfs:/path`, or other custom formats are not supported.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --diff --preserveMeta

Use --update

`--update`: Enables incremental synchronization. This option skips identical files and directories and synchronizes only new or changed files and directories from the source to the destination.

If a JindoDistCp task fails, you can use this parameter to resume it from the breakpoint and copy only the remaining files. You can also use this parameter to copy new files that are added to the source after the previous JindoDistCp task is completed.

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --update

Write data to OSS in Cold Archive, Archive, or IA storage classes

`--policy`: Specifies the storage class for data that is written to OSS. You can set this parameter to Cold Archive, Archive, or IA. If you do not specify this parameter, data is written to the Standard storage class by default.

• Write data to the Cold Archive storage class (coldArchive) in OSS

  This feature is available only in specific regions. For more information, see OSS storage classes. The following command provides an example:

   jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-bucket/hourly_table --policy coldArchive --parallelism 20

• Write data to the Archive storage class (archive) in OSS

   jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-bucket/hourly_table --policy archive --parallelism 20

• Write data to the IA storage class (ia) in OSS

   jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-bucket/hourly_table --policy ia --parallelism 20

Use --preserveMeta

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --preserveMeta

Use --jobBatch

 jindo-distcp-tool-${version}.jar --src /data/hourly_table --dest oss://example-oss-bucket/hourly_table --jobBatch 50000

Use --enableCMS

`--enableCMS`: Enables the CMS alert feature.

JindoDistCp counters

JindoDistCp counters summarize the results of a JindoDistCp task. The following table describes the counters.