Migrate data from HDFS to OSS-Object Storage Service(OSS)-阿里云帮助中心

This topic describes how to use Jindo DistCp to migrate data from the Hadoop Distributed File System (HDFS) to Object Storage Service (OSS), using an Alibaba Cloud E-MapReduce cluster as an example.

Background information

In traditional big data systems, HDFS often serves as the underlying storage for large-scale datasets. For data migration and copy tasks, the most common tool is the DistCp utility that comes with Hadoop. However, this tool cannot fully leverage the features of object storage, leading to poor performance and potential data consistency issues. It also offers a limited set of options that may not meet all user requirements.

Alibaba Cloud Jindo DistCp is a distributed file copy utility designed for copying files within or between large-scale clusters. It uses MapReduce to manage file distribution, error handling, and recovery. The job takes a list of files and directories as input, and each map task copies a subset of files from the source. Jindo DistCp fully supports data copy scenarios between HDFS clusters, between HDFS and OSS, and between OSS buckets. It also provides a wide range of parameters and strategies to customize copy tasks.

Jindo DistCp offers the following advantages over Hadoop DistCp for migrating data from HDFS to OSS:

High performance: Up to 1.59 times faster than Hadoop DistCp in test scenarios.
Rich features: Offers multiple copy methods and scenario-specific optimization strategies.
Deep integration with OSS: Supports native OSS features, such as changing a file's storage class to Archive or applying compression.
Atomic copy: Guarantees data consistency with a no-rename copy mechanism.
Broad compatibility: Serves as a full replacement for Hadoop DistCp and supports Hadoop 2.7+ and Hadoop 3.x.

Prerequisites

You must have an E-MapReduce cluster running version 3.28.0 or later. For more information, see Create a cluster.

In E-MapReduce clusters of version 3.28.0 or later, you can run Jindo DistCp by using shell commands. For more information, see Jindo DistCp usage instructions. For clusters with versions earlier than 3.28.0, you may encounter compatibility issues. If this occurs, submit a ticket for assistance.

If you use a self-managed cluster on ECS instances, you must have a Hadoop 2.7+ or Hadoop 3.x environment capable of running MapReduce jobs.

Step 1: Download the JAR package

Log on to the E-MapReduce cluster.
1. Log on to the EMR on ECS console.
2. Click the ID of the E-MapReduce cluster.
3. Click the Nodes tab, and then click the icon to the left of a node group.
4. Click an ECS ID. On the instance details page, click Connect to the right of the instance ID.
To connect to the cluster from a Windows or Linux environment by using an SSH key pair or password, see Log on to a cluster.
Download and decompress the latest version of Jindosdk-${version}.tar.gz. For the download link, see Download JindoData.

Step 2: Configure OSS credentials

Use one of the following methods to configure credentials for OSS access.

Command-line parameters

hadoop jar jindo-distcp-tool-${version}.jar --src /tmp/ --dest oss://examplebucket/ --hadoopConf fs.oss.accessKeyId=LTAI**************** --hadoopConf fs.oss.accessKeySecret=yourAccessKeySecret --hadoopConf fs.oss.endpoint=oss-cn-hangzhou.aliyuncs.com --parallelism 10

Configuration file

Navigate to the directory that contains the Hadoop core-site.xml configuration file.
```
cd /etc/emr/hadoop-conf/
```
For more information about common file paths in an E-MapReduce environment, see Common file paths.
Open the core-site.xml file.
```
vim core-site.xml
```

Add the following properties to the core-site.xml file.

<configuration>
    <property>
        <name>fs.oss.accessKeyId</name>
        <value>LTAI****************</value>
    </property>

    <property>
        <name>fs.oss.accessKeySecret</name>
        <value>YourAccessKeySecret</value>
    </property>

    <property>
        <name>fs.oss.endpoint</name>
        <!-- In an Alibaba Cloud ECS environment, we recommend that you use an internal OSS endpoint, such as oss-cn-xxx-internal.aliyuncs.com. -->
        <value>oss-cn-xxx.aliyuncs.com</value>
    </property>
</configuration>

Credential-free access

To avoid storing your AccessKey pair in plaintext and to improve security, configure credential-free access. For more information, see Use the credential-free feature of JindoFS SDK.

Step 3: Migrate or copy data

Run the following command to list the data in HDFS:

hdfs dfs -ls /

Sample response:

Found 8 items
drwxrwxrwx   - admin  supergroup          0 2023-10-26 10:55 /.sysinfo
drwxrwxrwx   - hadoop supergroup          0 2023-10-26 10:55 /apps
drwxrwxrwx   - root   supergroup          0 2022-08-03 15:54 /data
-rw-r-----   1 root   supergroup         13 2022-08-25 11:45 /examplefile.txt
drwxrwxrwx   - spark  supergroup          0 2023-10-26 14:49 /spark-history
drwx-wx-wx   - hive   supergroup          0 2023-10-26 13:35 /tmp
drwxrwxrwx   - hive   supergroup          0 2023-10-26 14:48 /user
drwxrwxrwx   - hadoop supergroup          0 2023-10-26 14:48 /yarn

Change to the directory where the jindo-distcp-tool-${version}.jar package is located.
```
cd /opt/apps/JINDOSDK/jindosdk-current/tools
```

Run a command to migrate data from HDFS to OSS.

Full data copy

Perform a full data migration or copy of data from the HDFS directory /tmp to the OSS destination path oss://examplebucket. The following is an example command:

hadoop jar jindo-distcp-tool-${version}.jar --src /tmp/ --dest oss://examplebucket/ --hadoopConf fs.oss.accessKeyId=LTAI**************** --hadoopConf fs.oss.accessKeySecret=yourAccessKeySecret --hadoopConf fs.oss.endpoint=oss-cn-hangzhou.aliyuncs.com --parallelism 10

The following table describes the parameters and options in the command.

Parameter	Description	Example
--src	The HDFS source path.	/tmp/
--dest	The destination OSS path.	oss://examplebucket/
--hadoopConf	Specifies the AccessKey ID, AccessKey secret, and endpoint required to access OSS. To obtain an AccessKey ID and AccessKey secret, see Create an AccessKey pair. For a list of OSS regions and their corresponding endpoints, see Regions and endpoints. Important If you are accessing OSS from an ECS instance, use an internal endpoint, such as oss-cn-xxx-internal.aliyuncs.com.	--hadoopConf fs.oss.accessKeyId=LTAI**************** --hadoopConf fs.oss.accessKeySecret=yourAccessKeySecret --hadoopConf fs.oss.endpoint=oss-cn-hangzhou.aliyuncs.com
--parallelism	The number of concurrent tasks. You can adjust this value based on your cluster's resources.	10

Incremental data copy

If you want to copy only the data that was added to the source path after the last full data migration or copy, you can use the --update option to perform an incremental data migration or copy.

The following example command incrementally migrates or copies data from the specified HDFS directory /tmp to the destination OSS path oss://examplebucket/.

hadoop jar jindo-distcp-tool-${version}.jar --src /tmp/ --dest oss://examplebucket/ --hadoopConf fs.oss.accessKeyId=LTAI**************** --hadoopConf fs.oss.accessKeySecret=yourAccessKeySecret --hadoopConf fs.oss.endpoint=oss-cn-hangzhou.aliyuncs.com --update --parallelism 10

When you use the --update option, checksum is enabled by default. When this feature is enabled, DistCp compares the file names, file sizes, and checksums between the source and destination paths. If the file name, file size, or checksum is inconsistent between the source and destination paths, an incremental data migration or a copy task is automatically triggered.

If you do not need to compare the checksums of files in the source and destination paths, add the --disableChecksum option to disable checksum verification. The example command is as follows:

hadoop jar jindo-distcp-tool-${version}.jar --src /tmp/ --dest oss://examplebucket/ --hadoopConf fs.oss.accessKeyId=LTAI**************** --hadoopConf fs.oss.accessKeySecret=yourAccessKeySecret --hadoopConf fs.oss.endpoint=oss-cn-hangzhou.aliyuncs.com --update --disableChecksum --parallelism 10

Appendix 1: Jindo DistCp parameters

Jindo DistCp provides a comprehensive set of parameters and options. To view detailed usage information for each one, run the following command:

hadoop jar jindo-distcp-tool-${version}.jar --help

The following table describes the common parameters and options.

Parameter	Description	Example
--src	The source path to copy from.	--src oss://examplebucket/sourcedir
--dest	The destination path to copy to.	--dest oss://examplebucket/destdir
--bandWidth	The bandwidth limit in MB/s per host for the DistCp job.	--bandWidth 6
--codec	Specifies the compression codec for the files. Supported codecs include `gzip`, `gz`, `lzo`, `lzop`, and `snappy`. You can also use the following keywords: `none`: Saves files as uncompressed. If a source file is compressed, Jindo DistCp decompresses it. `keep` (default): Preserves the original compression of the files. Note To use the `lzo` codec in an open source Hadoop environment, you must have the gplcompression native library and the hadoop-lzo package installed. If they are not available, we recommend that you use a different compression codec.	--codec gz
--policy	Specifies the storage class for the files copied to OSS. Valid values: ia: Infrequent Access archive: Archive coldArchive: Cold Archive	--policy coldArchive
--filters	Specifies a path to a file that contains regular expressions. Each line in the file is a pattern for files to exclude from the copy or comparison process.	--filters test.txt
--srcPrefixesFile	Specifies a file containing a list of file paths to copy. The paths in the file are relative to the `--src` path.	--srcPrefixesFile prefixes.txt
--parallelism	Specifies the value for the `mapreduce.job.reduces` parameter in the MapReduce job. You can customize this value to control the concurrency of the DistCp job based on your cluster's resources. Note The default value for this parameter in an E-MapReduce environment is 7.	--parallelism 20
--tmp	Specifies a temporary directory for the DistCp job.	--tmp /tmp
--hadoopConf	Specifies the AccessKey pair and endpoint for accessing OSS.	--hadoopConf fs.oss.accessKeyId=LTAI**************** --hadoopConf fs.oss.accessKeySecret=yourAccessKeySecret --hadoopConf fs.oss.endpoint=oss-cn-xxx.aliyuncs.com
--disableChecksum	Disables checksum verification.	--disableChecksum
--deleteOnSuccess	Deletes the source files after the copy job completes successfully.	--deleteOnSuccess
--enableTransaction	Guarantees job-level integrity and transactional support.	--enableTransaction
--ignore	Ignores exceptions that occur during data migration. Errors do not interrupt the job and are reported as DistCp counters. If `--enableCMS` is used, notifications are also sent through the specified channel.	-ignore
--diff	Verifies that all files were copied and lists any that were missed.	--diff
--update	Enables an incremental copy, which copies only new or updated files from the source path to the destination path.	--update
--preserveMeta	Preserves file metadata, including Owner, Group, Permission, Atime, Mtime, Replication, BlockSize, XAttrs, and ACLs, during the migration.	--preserveMeta
--jobBatch	Specifies the number of files processed by each DistCp job. Default value: 1000.	--jobBatch 1000
--taskBatch	Specifies the number of files processed by each DistCp subtask. Default value: 10.	--taskBatch 10

Appendix 2: Use cases

Scenario 1: Verify data integrity

Jindo DistCp provides two methods to verify data integrity.

Method 1: Use DistCp counters

You can verify data integrity by checking the DistCp counters, such as BYTES_EXPECTED and FILES_EXPECTED, in the job summary.

Example
    JindoDistcpCounter
        BYTES_COPIED=10000
        BYTES_EXPECTED=10000
        FILES_COPIED=11
        FILES_EXPECTED=11
        ...
    Shuffle Errors
        BAD_ID=0
        CONNECTION=0
        IO_ERROR=0
        WRONG_LENGTH=0
        WRONG_MAP=0
        WRONG_REDUCE=0

The following table describes the counters in the example.

Parameter	Description
BYTES_COPIED	The number of bytes successfully copied.
BYTES_EXPECTED	The total number of bytes expected to be copied.
FILES_COPIED	The number of files successfully copied.
FILES_EXPECTED	The total number of files expected to be copied.
FILES_SKIPPED	The number of files skipped during an incremental copy.
BYTES_SKIPPED	The number of bytes skipped during an incremental copy.
COPY_FAILED	The number of files that failed to copy. An alert is triggered if this value is not 0.
BYTES_FAILED	The number of bytes that failed to copy.
DIFF_FILES	The number of files that are different between the source and destination paths. An alert is triggered if this value is not 0.
DIFF_FAILED	The number of files that caused an exception during the comparison operation. This count is included in `DIFF_FILES`.
SRC_MISS	The number of files missing from the source path. This count is included in `DIFF_FILES`.
DST_MISS	The number of files missing from the destination path. This count is included in `DIFF_FILES`.
LENGTH_DIFF	The number of files with mismatched sizes between the source and destination. This count is included in `DIFF_FILES`.
CHECKSUM_DIFF	The number of files that failed the checksum verification. This count is included in `COPY_FAILED`.
SAME_FILES	The number of files that are identical in both the source and destination paths.

Method 2: Use the --diff option

In the example, you can use the --diff option to compare the file names and file sizes between the source path and the destination path.

hadoop jar jindo-distcp-tool-${version}.jar --src /tmp/ --dest oss://examplebucket/ --hadoopConf fs.oss.accessKeyId=LTAI**************** --hadoopConf fs.oss.accessKeySecret=yourAccessKeySecret --hadoopConf fs.oss.endpoint=oss-cn-hangzhou.aliyuncs.com --diff

Scenario 2: Specify a storage class

You can add the --policy option in the example to specify the storage class for files written to OSS. The following example shows how to specify the storage class as Infrequent Access:

hadoop jar jindo-distcp-tool-${version}.jar --src /tmp/ --dest oss://examplebucket/ --hadoopConf fs.oss.accessKeyId=LTAI**************** --hadoopConf fs.oss.accessKeySecret=yourAccessKeySecret --hadoopConf fs.oss.endpoint=oss-cn-hangzhou.aliyuncs.com --policy ia --parallelism 10

If you want to specify the storage type as Archive, replace --policy ia with --policy archive. If you want to specify the storage type as Cold Archive, replace --policy ia with --policy coldArchive. In addition, Cold Archive is only available in select regions. For more information, see Cold Archive.

Scenario 3: Delete source data after migration

You can use the --deleteOnSuccess option to retain only the data in the OSS destination path and delete the specified data from the HDFS source path after a migration or copy task is completed.

hadoop jar jindo-distcp-tool-${version}.jar --src /tmp/ --dest oss://examplebucket/ --hadoopConf fs.oss.accessKeyId=LTAI**************** --hadoopConf fs.oss.accessKeySecret=yourAccessKeySecret --hadoopConf fs.oss.endpoint=oss-cn-hangzhou.aliyuncs.com --deleteOnSuccess --parallelism 10