E-MapReduce:Use Jindo DistCp in specific scenarios

Common use scenarios of Jindo DistCp:

• Scenario 1: What parameters are required for optimization if I import a large amount of HDFS data or a large number of HDFS files (millions or tens of millions) to OSS?
• Scenario 2: How do I verify data integrity after I use Jindo DistCp to import data to OSS?
• Scenario 3: What parameters are required to support a resumable upload if a DistCp task fails to import HDFS data to OSS?
• Scenario 4: What parameters are required to process the files that may be generated when I use Jindo DistCp to import HDFS data to OSS?
• Scenario 5: What parameters are required to specify the YARN queue where a Jindo DistCp task resides and the available bandwidth allocated to the task?
• Scenario 6: What parameters are required when I write data to OSS Cold Archive, Archive, or IA storage?
• Scenario 7: What parameters are required to accelerate file transfer based on the proportion of small files and specific file size?
• Scenario 8: What parameters are required if Amazon S3 is used as a data source?
• Scenario 9: What parameters are required if I want to copy files to OSS and compress the copied files in the LZO or GZ format?
• Scenario 10: What parameters are required if I want to copy the files that meet specific rules or the files in some sub-directories of the same parent directory?
• Scenario 11: What parameters are required if I want to merge the files that meet specific rules to reduce the number of files?
• Scenario 12: What parameters are required if I want to delete original files after a copy operation?
• Scenario 13: What do I do if I do not want to specify AccessKey pair and endpoint information of OSS in the CLI?

If you are not using EMR, the following conditions must be met before you can import HDFS data to Object Storage Service (OSS):

• You can read data from HDFS.
• The AccessKey ID, AccessKey secret, and endpoint of OSS are obtained. You can write data to a destination OSS bucket.
• The storage class of the OSS bucket is not Archive.
• You can submit MapReduce tasks.
• The JAR package of Jindo DistCp is downloaded.

Example:

hadoop jar jindo-distcp-<version>.jar --src /data/incoming/hourly_table --dest oss://destBucket/hourly_table --ossKey yourkey --ossSecret yoursecret --ossEndPoint oss-cn-hangzhou.aliyuncs.com --parallelism 10

To import a large amount of data or a large number of files, such as millions or tens of millions of files, to OSS, you can set parallelism to a large value to increase the parallelism. You can also set enableBatch to true for optimization. Optimization command:

hadoop jar jindo-distcp-<version>.jar --src /data/incoming/hourly_table --dest oss://destBucket/hourly_table --ossKey yourkey --ossSecret yoursecret --ossEndPoint oss-cn-hangzhou.aliyuncs.com --parallelism 500 --enableBatch

You can use one of the following methods to verify data integrity:

• Jindo DistCp Counters
  Check DistCp Counters in the counter information of a MapReduce task.

  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

  where:

  • Bytes Destination Copied specifies the size of the file that you copy to a destination directory, in bytes.
  • Bytes Source Read specifies the size of the file that you read from the source directory, in bytes.
  • Files Copied specifies the number of copied files.
• Jindo DistCp --diff

  Run a command that ends with the --diff parameter to compare the files in the source and destination directories. This command compares the names and sizes of the files in the source and destination directories, identifies the files that are not copied or failed to be copied, and generates a manifest file in the directory to which the command is submitted.

  Add --diff to the command described in Scenario 1. Example:

  hadoop jar jindo-distcp-<version>.jar --src /data/incoming/hourly_table --dest oss://destBucket/hourly_table --ossKey yourkey --ossSecret yoursecret --ossEndPoint oss-cn-hangzhou.aliyuncs.com --diff

  If all files are copied, the following information is returned:

  INFO distcp.JindoDistCp: distcp has been done completely

If your DistCp task fails and you want to perform resumable upload to copy only the files that failed to be copied, perform the following operations on the basis of the command described in Scenario 1:

1. Add --diff to check whether all files are copied.

   hadoop jar jindo-distcp-<version>.jar --src /data/incoming/hourly_table --dest oss://destBucket/hourly_table --ossKey yourkey --ossSecret yoursecret --ossEndPoint oss-cn-hangzhou.aliyuncs.com --diff

   If all the files are copied, the following information is returned.

   INFO distcp.JindoDistCp: distcp has been done completely.

2. If some files are not copied, a manifest file is generated. In this case, use --copyFromManifest and --previousManifest to copy the remaining files. Example:

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

   file:///opt/manifest-2020-04-17.gz is the path where the manifest file is stored.

Add --update to the command described in Scenario 1 to copy incremental files.

hadoop jar jindo-distcp-<version>.jar --src /data/incoming/hourly_table --dest oss://destBucket/hourly_table --ossKey yourkey --ossSecret yoursecret --ossEndPoint oss-cn-hangzhou.aliyuncs.com --parallelism 10 --update

Add the following two parameters to the command described in Scenario 1. These two parameters can be used together or separately.

• --queue: the name of the YARN queue.
• --bandwidth: the size of the specified bandwidth, in MB/s.

Example:

hadoop jar jindo-distcp-<version>.jar --src /data/incoming/hourly_table --dest oss://destBucket/hourly_table --ossKey yourkey --ossSecret yoursecret --ossEndPoint oss-cn-hangzhou.aliyuncs.com --queue yarnqueue --bandwidth 6 --parallelism 10

Replace the parameters that specify the AccessKey ID, AccessKey secret, and endpoint of OSS in the command described in Scenario 1 with the following parameters of Amazon S3:

• --s3Key: the AccessKey ID for Amazon S3
• --s3Secret: the AccessKey secret for Amazon S3
• --s3EndPoint: the endpoint for Amazon S3

Example:

hadoop jar jindo-distcp-<version>.jar --src s3a://yourbucket/ --dest oss://destBucket/hourly_table --s3Key yourkey --s3Secret yoursecret --s3EndPoint s3-us-west-1.amazonaws.com --parallelism 10

Add --outputCodec to the command described in Scenario 1. Example:

hadoop jar jindo-distcp-<version>.jar --src /data/incoming/hourly_table --dest oss://destBucket/hourly_table --ossKey yourkey --ossSecret yoursecret --ossEndPoint oss-cn-hangzhou.aliyuncs.com --outputCodec=gz --parallelism 10

You can set --outputCodec to gzip, gz, lzo, lzop, snappy, none, or keep. Default value: keep. Descriptions of none and keep:

• none: Jindo DistCp does not compress copied files. If the files have been compressed, Jindo DistCp decompresses them.
• keep: Jindo DistCp copies files with no change to the compression.

Add the following parameters to the command described in Scenario 1:

• --targetSize: the maximum size of the merged file, in MB
• --groupBy: the merging rule, which is a regular expression

Example:

hadoop jar jindo-distcp-<version>.jar --src /data/incoming/hourly_table --dest oss://destBucket/hourly_table --ossKey yourkey --ossSecret yoursecret --ossEndPoint oss-cn-hangzhou.aliyuncs.com --targetSize=10 --groupBy='.*/([a-z]+).*.txt' --parallelism 20

Add --deleteOnSuccess to the command described in Scenario 1. Example:

hadoop jar jindo-distcp-<version>.jar --src /data/incoming/hourly_table --dest oss://destBucket/hourly_table --ossKey yourkey --ossSecret yoursecret --ossEndPoint oss-cn-hangzhou.aliyuncs.com --deleteOnSuccess --parallelism 10

Jindo DistCp allows you to save the AccessKey ID, AccessKey secret, and endpoint of OSS in the core-site.xml file so that you do not need to repeatedly specify the information.

• If you want to save the AccessKey ID, AccessKey secret, and endpoint of OSS, save the following information into the core-site.xml file:

  <configuration>
      <property>
          <name>fs.jfs.cache.oss-accessKeyId</name>
          <value>xxx</value>
      </property>
  
      <property>
          <name>fs.jfs.cache.oss-accessKeySecret</name>
          <value>xxx</value>
      </property>
  
      <property>
          <name>fs.jfs.cache.oss-endpoint</name>
          <value>oss-cn-xxx.aliyuncs.com</value>
      </property>
  </configuration>

• If you want to save the AccessKey ID, AccessKey secret, and endpoint of Amazon S3, save the following information into the 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>