Jindo DistCp の使用

このトピックでは、データコピーツール Jindo DistCp の使用方法について説明します。

前提条件

Java Development Kit ( JDK ) 8 がコンピューターにインストールされていること。
バージョン 3.28.0 以降の E-MapReduce ( EMR ) クラスターが作成されていること。クラスターの作成方法については、クラスターの作成をご参照ください。

SSH モードで EMR クラスターのマスターノードにログオンします。
詳細については、クラスターへのログオンをご参照ください。

ヘルプ情報を取得するには、次のコマンドを実行します。

jindo distcp --help

次の情報が返されます。

     --help           - ヘルプテキストを表示します
     --src=VALUE          - ファイルのコピー元ディレクトリ
     --dest=VALUE              - ファイルのコピー先ディレクトリ
     --parallelism=VALUE         - コピー タスクの並列度
     --outputManifest=VALUE       - マニフェスト ファイルの名前
     --previousManifest=VALUE   -   既存のマニフェスト ファイルへのパス
     --requirePreviousManifest=VALUE   -   指定されている場合、以前のマニフェストが存在する必要があります
     --copyFromManifest   -   ディレクトリをリストする代わりに、マニフェストからコピーします
     --srcPrefixesFile=VALUE   -   コピー元 URI プレフィックスのリストを含むファイル
     --srcPattern=VALUE   -   このパターンに一致するコピー元ファイルのみを含めます
     --deleteOnSuccess   -   コピーが成功したら入力ファイルを削除します
     --outputCodec=VALUE   -   出力ファイルの圧縮コーデック
     --groupBy=VALUE   -   入力ファイルをグループ化するパターン
     --targetSize=VALUE   -   出力ファイルのターゲット サイズ
     --enableBalancePlan   -   バランスをとるためのプランコピー タスクを有効にします
     --enableDynamicPlan   -   プランコピー タスクを動的に有効にします
     --enableTransaction   -   ジョブのトランザクションを明示的に有効にします
     --diff   -   コピー元とコピー先のファイルリストの差分を表示します
     --ossKey=VALUE   -   必要に応じて OSS キーを指定します
     --ossSecret=VALUE   -   必要に応じて OSS シークレットを指定します
     --ossEndPoint=VALUE   -   必要に応じて OSS エンドポイントを指定します
     --policy=VALUE   -   OSS ストレージ ポリシーを指定します
     --cleanUpPending   -   distcp ジョブの完了時に、不完全なアップロードをクリーンアップします
     --queue=VALUE   -   必要に応じて yarn キュー名を指定します
     --bandwidth=VALUE   -   必要に応じて、マップ/リデュースごとの帯域幅を MB 単位で指定します
     --s3Key=VALUE   -   S3 キーを指定します
     --s3Secret=VALUE   -   S3 シークレットを指定します
     --s3EndPoint=VALUE   -   S3 エンドポイントを指定します

--src と --dest

--src はコピー元ディレクトリを指定します。--dest はコピー先ディレクトリを指定します。

デフォルトでは、Jindo DistCp は --src で指定されたディレクトリ内のすべてのファイルを --dest で指定されたディレクトリにコピーします。ルートディレクトリを指定しない場合、Jindo DistCp は自動的にルートディレクトリを作成します。

たとえば、次のコマンドを実行して、HDFS の /opt/tmp ディレクトリにあるファイルを OSS バケットにコピーできます。

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

説明

yourBucketName は OSS バケットの名前です。

--parallelism

--parallelism は、ファイルのコピーを実行するために実行される MapReduce ジョブの mapreduce.job.reduces パラメーターを指定します。デフォルト値は 7 です。使用可能なクラスターリソースに基づいて --parallelism をカスタマイズできます。これにより、いくつのリデュースタスクを並列して実行できるかを決定できます。

たとえば、次のコマンドを実行して、HDFS の /opt/tmp ディレクトリにあるファイルを OSS バケットにコピーできます。

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

--srcPattern

--srcPattern は、コピー操作のファイルをフィルタリングする正規表現を指定します。正規表現は完全なパスと一致する必要があります。

たとえば、/data/incoming/hourly_table/2017-02-01/03 ディレクトリにあるすべてのログファイルをコピーする必要がある場合は、--srcPattern を .*\.log に設定します。

次のコマンドを実行して、/data/incoming/hourly_table/2017-02-01/03 ディレクトリ内のファイルを表示します。

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

次の情報が返されます。

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

ログファイルをコピーするには、次のコマンドを実行します。

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

次のコマンドを実行して、コピー先の OSS バケット内のファイルを表示します。

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

次の情報が返されます。コピー元ディレクトリにあるログファイルのみがコピーされます。

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

--deleteOnSuccess

--deleteOnSuccess を使用すると、Jindo DistCp はコピー操作が成功した後に、コピーされたファイルをコピー元ディレクトリから削除できます。

たとえば、次のコマンドを実行して、/data/incoming/hourly_table 内のファイルを OSS バケットにコピーし、コピーされたファイルをコピー元ディレクトリから削除できます。

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

--outputCodec

--outputCodec は、コピーされたファイルをオンラインで圧縮するために使用される圧縮コーデックを指定します。例:

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

次のコマンドを実行して、コピー先ディレクトリ内のファイルを表示します。

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

次の情報が返されます。コピー先ディレクトリ内のファイルは GZ 形式で圧縮されています。

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

このパラメーターは、gzip、gz、lzo、lzop、snappy、none、または keep に設定できます。デフォルト値: keep。none と keep の説明:

none: Jindo DistCp はコピーされたファイルを圧縮しません。ファイルが圧縮されている場合、Jindo DistCp はファイルを解凍します。
keep: Jindo DistCp は、圧縮を変更せずにファイルをコピーします。

説明

オープンソースの Hadoop クラスターで LZO コーデックを使用する場合は、gplcompression のネイティブライブラリと hadoop-lzo パッケージをインストールする必要があります。

--outputManifest と --requirePreviousManifest

--outputManifest は、Jindo DistCp によってコピーされたすべてのファイルに関する情報を含むマニフェストファイルを生成します。情報には、コピー先ファイル、コピー元ファイル、およびファイルサイズが含まれます。

マニフェストファイルを生成する場合は、--requirePreviousManifest を false に設定します。デフォルトでは、ファイルは GZ 形式で圧縮されます。これが唯一サポートされている形式です。

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

次のコマンドを実行して、ファイルの内容を表示します。

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

次の情報が返されます。

{"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}
{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/2.log","baseName":"2017-02-01/03/2.log","srcDir":"oss://<yourBucketName>/hourly_table","size":4891}
{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/OPTIONS-000109","baseName":"2017-02-01/03/OPTIONS-000109","srcDir":"oss://<yourBucketName>/hourly_table","size":4891}
{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/emp01.txt","baseName":"2017-02-01/03/emp01.txt","srcDir":"oss://<yourBucketName>/hourly_table","size":1016}
{"path":"oss://<yourBucketName>/hourly_table/2017-02-01/03/emp06.txt","baseName":"2017-02-01/03/emp06.txt","srcDir":"oss://<yourBucketName>/hourly_table","size":1016}

--outputManifest と --previousManifest

--outputManifest は、以前と新しくコピーされたファイルの両方のリストを含むマニフェストファイルを生成します。--previousManifest は、以前にコピーされたファイルのリストを含むマニフェストファイルを生成します。このようにして、操作の全履歴を再作成し、現在のジョブによってコピーされたファイルを確認できます。

たとえば、コピー元ディレクトリに 2 つのファイルが追加されています。新しく追加されたファイルをコピーするには、次のコマンドを実行します。

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

コピーされたファイルを表示するには、次のコマンドを実行します。

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

次の情報が返されます。

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}

--copyFromManifest

--copyFromManifest を使用して、以前に --outputManifest によって生成されたマニフェストファイルを指定し、マニフェストファイルにリストされているファイルをコピー先ディレクトリにコピーできます。例:

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

--srcPrefixesFile

--srcPrefixesFile を使用すると、Jindo DistCp は複数のフォルダー内のファイルを一度にコピーできます。

次のコマンドを実行して、hourly_table の下のサブフォルダーを表示します。

hdfs dfs -ls oss://<yourBucketName>/hourly_table

次の情報が返されます。

Found 4 items
drwxrwxrwx   -          0 1970-01-01 08:00 oss://<yourBucketName>/hourly_table/2017-02-01
drwxrwxrwx   -          0 1970-01-01 08:00 oss://<yourBucketName>/hourly_table/2017-02-02

次のコマンドを実行して、hourly_table の下のすべてのファイルをコピー先フォルダーにコピーします。folders.txt という名前のファイルが生成されます。これには、hourly_table の下の各サブフォルダーに対応するプレフィックスが含まれています。

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

次のコマンドを実行して、folders.txt ファイルの内容を表示します。

cat 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

--groupBy と --targetSize

HDFS から多数の小さなファイルを読み取ると、データ処理のパフォーマンスに影響します。そのため、Jindo DistCp を使用して小さなファイルを指定されたサイズの大きなファイルにマージすることをお勧めします。これにより、分析パフォーマンスが最適化され、コストが削減されます。

次のコマンドを実行して、指定されたフォルダー内のファイルを表示します。

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

次の情報が返されます。

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

次のコマンドを実行して、フォルダー内の TXT ファイルをそれぞれ 10 MB 以下のファイルにマージします。

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

コピー先ディレクトリ内のファイルを表示するには、次のコマンドを実行します。2 つの TXT ファイルが 1 つにマージされます。

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

--enableBalancePlan

小さいファイルと大きいファイルの両方をコピーする必要があるが、小さいファイル間と大きいファイル間でファイルサイズに大きな違いがない場合は、--enableBalancePlan を使用してジョブ割り当てプランを最適化できます。これにより、Jindo DistCp のコピーパフォーマンスが向上します。アプリケーションプランを指定しない場合、ファイルはランダムに割り当てられます。

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

説明

このオプションは、--groupBy または --targetSize と同じコマンドで使用することはできません。

--enableDynamicPlan

コピーするファイルのサイズが大きく異なり、ほとんどのファイルが小さなファイルである場合は、--enableDynamicPlan を使用してジョブ割り当てプランを最適化できます。これにより、Jindo DistCp のコピーパフォーマンスが向上します。

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

説明

このオプションは、--groupBy または --targetSize と同じコマンドで使用することはできません。

--enableTransaction

--enableTransaction は、ジョブレベルの整合性とジョブ間のトランザクションサポートを保証します。例:

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

--diff

ファイルのコピー後、--diff を使用して、コピー元ディレクトリとコピー先ディレクトリの差分を確認できます。

例:

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

すべてのファイルがコピーされている場合、次の情報が返されます。

INFO distcp.JindoDistCp: distcp has been done completely

一部のファイルがコピーされていない場合は、これらのファイルのリストを含む manifest ファイルがコピー先ディレクトリに生成されます。次に、--copyFromManifest と --previousManifest を使用して、リスト内のファイルをコピー先ディレクトリにコピーできます。このようにして、データ量とファイル数が検証されます。Jindo DistCp がコピー処理中に圧縮または解凍操作を実行した場合、--diff は正確なファイルサイズの差分を返しません。

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

説明

コピー先ディレクトリが HDFS ディレクトリの場合、--dest を /path、hdfs://hostname:ip/path、または hdfs://headerIp:ip/path の形式で指定する必要があります。hdfs:///path や hdfs:/path などの他の形式はサポートされていません。

--queue

--queue は、現在の DistCp タスクが存在する YARN キューの名前を指定します。

例:

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

--bandwidth

--bandwidth は、現在の DistCp タスクに割り当てられた帯域幅を指定します。これにより、タスクが過剰な帯域幅を使用するのを防ぎます。単位: MB/s。

AccessKey ペアを使用して OSS にアクセスする

EMR の外部のインスタンスから OSS にアクセスする場合、または AccessKey を使用しないアクセスがサポートされていない場合は、AccessKey ペアを使用して OSS にアクセスできます。コマンドで --key、--secret、および --endPoint パラメーターを設定して、AccessKey ペアを指定します。

例:

jindo distcp --src /data/incoming/hourly_table --dest oss://<yourBucketName>/hourly_table --key <yourAccessKeyId> --secret <yourAccessKeySecret> --endPoint oss-cn-hangzhou.aliyuncs.com --parallelism 20

yourAccessKeyId は、Alibaba Cloud アカウントの AccessKey ID です。yourAccessKeySecret は、Alibaba Cloud アカウントの AccessKey シークレットです。

OSS IA またはアーカイブストレージにデータを書き込む

DistCp タスクを使用して OSS にデータを書き込む場合、OSS アーカイブまたは低頻度アクセス ( IA ) ストレージにデータを書き込むことができます。

アーカイブストレージクラスの例 ( --archive ) :

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

IA ストレージクラスの例 ( --ia ) :

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

残留ファイルをクリーンアップする

DistCp タスクを実行すると、コピー先ディレクトリに正しくアップロードされていないファイルが生成される場合があります。ファイルは uploadId に基づいて OSS によって管理され、ユーザーには見えない場合があります。この場合、コマンドで --cleanUpPending パラメーターを指定できます。このようにして、タスクの完了後にシステムによって残留ファイルが自動的にクリーンアップされます。または、OSS コンソールでファイルをクリーンアップすることもできます。

例:

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

Amazon S3 をデータソースとして使用する

コマンドで --s3Key、--s3Secret、および --s3EndPoint パラメーターを使用して、Amazon S3 関連の情報を指定できます。

例:

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

Hadoop の core-site.xml ファイルで s3Key、s3Secret、および s3EndPoint パラメーターを設定できます。このようにして、毎回 AccessKey ペアを指定する必要はありません。

<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>

例:

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

DistCp カウンターを確認する

MapReduce のカウンター情報で DistCp カウンターを確認するには、次のコマンドを実行します。

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

説明

Jindo DistCp がコピー処理中に圧縮または解凍操作を実行した場合、Bytes Destination Copied と Bytes Source Read の値が異なる場合があります。

前提条件