Jindo DistCp の使用 - E-MapReduce - Alibaba Cloud ドキュメントセンター

Jindo DistCp は、MapReduce 上に構築された分散データコピーツールです。これを使用して、Hadoop 分散ファイルシステム (HDFS) と Object Storage Service (OSS) 間、OSS バケット間、または Amazon S3 から OSS へファイルをコピーできます。フィルター、圧縮、増分コピー、パフォーマンスチューニングのオプションも利用できます。

前提条件

開始する前に、以下の要件が満たされていることを確認してください。

コマンドを実行するマシンに Java Development Kit (JDK) 8 がインストールされていること
バージョン 3.28.0 以降の E-MapReduce (EMR) クラスター (クラスターの作成)

パラメーター

jindo distcp --help を EMR マスターノードで実行すると、利用可能なすべてのパラメーターが表示されます。

以下の表に、各パラメーター、必須かどうか、およびその機能を示します。

パラメーター	必須	説明
`--src`	必須	コピー元のディレクトリ
`--dest`	必須	コピー先のディレクトリ
`--parallelism`	任意	並列 Reduce タスクの数。`mapreduce.job.reduces` に対応します。デフォルト： `7`
`--srcPattern`	任意	ソースファイルをフィルターするための正規表現。フルパスと一致する必要があります。
`--deleteOnSuccess`	任意	コピーが成功した後にソースファイルを削除します。
`--outputCodec`	任意	コピー先のファイルに適用する圧縮コーデック。値： `gzip`、 `gz`、 `lzo`、 `lzop`、 `snappy`、 `none`、 `keep`。デフォルト： `keep`
`--outputManifest`	任意	生成するマニフェストファイルの名前。コピーされたすべてのファイル (コピー先パス、ソースパス、ファイルサイズ) の一覧が含まれます。gzip 形式のみがサポートされています。
`--requirePreviousManifest`	任意	`false` に設定すると、以前のマニフェストファイルを必須とせずにマニフェストファイルを生成します。
`--previousManifest`	任意	以前にコピーされたファイルを記載した既存のマニフェストファイルへのパス
`--copyFromManifest`	任意	ディレクトリをスキャンする代わりに、マニフェストファイルにリストされているファイルをコピーします。
`--srcPrefixesFile`	任意	ソース URI プレフィックスを含むファイルへのパス (1 行に 1 つ)。1 つのジョブで複数のディレクトリからファイルをコピーします。
`--groupBy`	任意	入力ファイルをグループ化してマージするための正規表現パターン
`--targetSize`	任意	マージされた出力ファイルのターゲットサイズ (MB 単位)
`--enableBalancePlan`	任意	ファイルサイズが類似している場合にタスク割り当てを最適化します。`--groupBy` または `--targetSize` とは組み合わせることはできません。
`--enableDynamicPlan`	任意	ほとんどのファイルが小さく、ファイルサイズが大幅に異なる場合にタスク割り当てを最適化します。`--groupBy` または `--targetSize` とは組み合わせることはできません。
`--enableTransaction`	任意	整合性を確保するために、ジョブレベルのトランザクションサポートを有効にします。
`--diff`	任意	ソースとコピー先のファイル一覧を比較し、差異を報告します。
`--ossKey`	任意	OSS アクセス用の AccessKey ID
`--ossSecret`	任意	OSS アクセス用の AccessKey Secret
`--ossEndPoint`	任意	OSS エンドポイント
`--policy`	任意	書き込むデータに対する OSS ストレージポリシー。値： `archive`、 `ia` (低頻度アクセス)
`--cleanUpPending`	任意	ジョブの完了後、未完了のマルチパートアップロード (アップロード ID で識別) をクリーンアップします。
`--queue`	任意	コピージョブ用の YARN キュー名
`--bandwidth`	任意	map/reduce タスクあたりの帯域幅制限 (MB/s)
`--s3Key`	任意	Amazon S3 アクセスキー
`--s3Secret`	任意	Amazon S3 シークレットキー
`--s3EndPoint`	任意	Amazon S3 エンドポイント

制限事項

--enableBalancePlan と --enableDynamicPlan は、 --groupBy または --targetSize とは組み合わせることはできません。
--outputManifest では、マニフェストファイルとして gzip 形式のみがサポートされます。
コピー中に圧縮または展開が適用された場合、 --diff は正確なファイルサイズの差異を報告しません。
コピー先が HDFS ディレクトリの場合、 --dest は /path、 hdfs://hostname:port/path、または hdfs://headerIp:port/path のいずれかの形式を使用する必要があります。hdfs:///path および hdfs:/path の形式はサポートされていません。
オープンソースの Hadoop クラスターで LZO コーデックを使用するには、ネイティブの gplcompression ライブラリと hadoop-lzo パッケージをインストールする必要があります。

HDFS から OSS へのファイルのコピー

SSH 経由で EMR クラスターのマスターノードにログインし (SSH モードでのマスターノードへの接続)、そこからコマンドを実行します。

基本的なコピーコマンドは、ソースディレクトリ内のすべてのファイルをコピー先にコピーします：

jindo distcp --src /opt/tmp --dest oss://yang-hhht/tmp

コピー先のディレクトリが存在しない場合、Jindo DistCp は自動的に作成します。

コピーのスループットを向上させるには、 --parallelism をデフォルトの 7 より大きい値に設定します：

jindo distcp --src /opt/tmp --dest oss://yang-hhht/tmp --parallelism 20

パターンによるファイルのフィルタリング

--srcPattern に正規表現を使用すると、一致するファイルのみをコピーできます。パターンはファイルのフルパスと一致する必要があります。

たとえば、 /data/incoming/hourly_table から .log ファイルのみをコピーするには、次のようにします：

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --srcPattern '.*\.log' --parallelism 20

確認するには、まずソースディレクトリを一覧表示します：

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

コピー後、 2 つの .log ファイルのみがコピー先に表示されます：

hdfs dfs -ls oss://yang-hhht/hourly_table/2017-02-01/03

出力：

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

コピー後のソースファイルの削除

--deleteOnSuccess を追加すると、ファイルがコピー先に正常にコピーされた後にソースファイルが削除されます：

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --deleteOnSuccess --parallelism 20

コピー中のファイルの圧縮

--outputCodec を使用すると、コピー中にファイルをその場で圧縮できます。サポートされている値： gzip、 gz、 lzo、 lzop、 snappy、 none、 keep (デフォルト)。

keep：圧縮を変更せずにファイルをコピーします。
none：ファイルを非圧縮でコピーします。ファイルがすでに圧縮されている場合、展開されます。

gzip 圧縮を使用する例：

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --outputCodec=gz --parallelism 20

コピー先を一覧表示して、すべてのファイルが圧縮されていることを確認します：

hdfs dfs -ls oss://yang-hhht/hourly_table/2017-02-01/03

出力：

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

マニフェストファイルを使用した増分コピーの実行

マニフェストファイルには、ジョブによってコピーされたすべてのファイル (コピー先パス、ソースパス、ファイルサイズを含む) が記録されます。マニフェストを使用してコピー履歴を追跡し、新しいファイルのみをコピーする増分ジョブを実行します。

初回コピー用のマニフェストの生成

--requirePreviousManifest=false を設定すると、以前のジョブのマニフェストを必須とせずにマニフェストを作成できます。マニフェストは gzip 形式で保存されます。

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

マニフェストの内容を確認するには：

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

出力：

{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/000151.sst","baseName":"2017-02-01/03/000151.sst","srcDir":"oss://yang-hhht/hourly_table","size":2252}
{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/1.log","baseName":"2017-02-01/03/1.log","srcDir":"oss://yang-hhht/hourly_table","size":4891}
{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/2.log","baseName":"2017-02-01/03/2.log","srcDir":"oss://yang-hhht/hourly_table","size":4891}
{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/OPTIONS-000109","baseName":"2017-02-01/03/OPTIONS-000109","srcDir":"oss://yang-hhht/hourly_table","size":4891}
{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/emp01.txt","baseName":"2017-02-01/03/emp01.txt","srcDir":"oss://yang-hhht/hourly_table","size":1016}
{"path":"oss://yang-hhht/hourly_table/2017-02-01/03/emp06.txt","baseName":"2017-02-01/03/emp06.txt","srcDir":"oss://yang-hhht/hourly_table","size":1016}

以前のマニフェストを使用した新しいファイルのみのコピー

--previousManifest で以前のマニフェストを渡し、--outputManifest で新しいマニフェストを生成します。Jindo DistCp は、以前のマニフェストにすでにリストされているファイルをスキップし、新しいファイルのみをコピーします。新しいマニフェストには、完全なコピー履歴が含まれます。

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

このジョブで何が追加されたかを確認するには、 2 つのマニフェストの差分を取ります：

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

出力 — 2 つの新しいファイルがコピーされました：

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

マニフェストからのファイルの再コピー

--copyFromManifest と --previousManifest を組み合わせて使用し、マニフェストにリストされているファイルを正確にコピーします：

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

複数のソースディレクトリからのファイルのコピー

--srcPrefixesFile を使用すると、単一のジョブで複数のディレクトリからファイルをコピーできます。1 行に 1 つのソース URI プレフィックスを記載したプレーンテキストファイルを作成し、それをコマンドに渡します。

たとえば、次の内容の 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

次のコマンドを実行します：

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --srcPrefixesFile file:///opt/folders.txt --parallelism 20

小さいファイルのマージ

HDFS から多数の小さいファイルを読み取るのは低速です。--groupBy と --targetSize を組み合わせて使用し、コピー中に小さいファイルをより大きな出力ファイルにマージします。

--groupBy：入力ファイルをグループ化する正規表現。同じグループに一致するファイルは、1 つの出力ファイルにマージされます。
--targetSize：マージされた各出力ファイルの最大サイズ (MB 単位)。

たとえば、 hourly_table 内のすべての .txt ファイルをそれぞれ最大 10 MB のファイルにマージするには、次のようにします：

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

ソースディレクトリ (8 ファイル)：

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

コピー先ディレクトリ — 2 つの .txt ファイルが 1 つにマージされます：

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

コピーパフォーマンスのチューニング

混合ファイルサイズに対するタスク割り当てのバランス調整

ソースに小さいファイルと大きいファイルの両方が含まれており、各グループ内のサイズ差が大きくない場合は、 --enableBalancePlan を使用して、タスクを Reducer 間でより均等に分散します：

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --enableBalancePlan --parallelism 20

--groupBy または --targetSize とは組み合わせることはできません。

主にファイルサイズが小さく、サイズ分散が大きい場合の最適化

ほとんどのファイルが小さく、サイズが大幅に異なる場合は、 --enableDynamicPlan を使用してタスクを動的に割り当てます：

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --enableDynamicPlan --parallelism 20

--groupBy または --targetSize とは組み合わせることはできません。

帯域幅使用量の制限

--bandwidth を使用して、コピージョブが過剰なネットワーク帯域幅を消費するのを防ぎます。値は MB/s 単位で、map/reduce タスクごとに適用されます：

jindo distcp --src /data/incoming/hourly_table --dest oss://<your_bucket>/hourly_table --bandwidth 100

特定の YARN キューへの割り当て

--queue を使用して、ジョブを特定の YARN キューにルーティングします：

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

トランザクションサポートの有効化

--enableTransaction を使用して、ジョブレベルの整合性とジョブ間のトランザクションサポートを確保します：

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --enableTransaction --parallelism 20

コピーの完全性の検証

コピージョブの後、 --diff を使用してソースとコピー先のファイル一覧を比較します：

jindo distcp --src /data/incoming/hourly_table --dest oss://yang-hhht/hourly_table --diff

すべてのファイルが正常にコピーされた場合：

INFO distcp.JindoDistCp: distcp has been done completely

一部のファイルがコピーされなかった場合、 Jindo DistCp はコピー先ディレクトリに不足しているファイルを記載したマニフェストファイルを生成します。--copyFromManifest と --previousManifest を使用して、残りのファイルをコピーします：

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

コピー中に圧縮または展開が適用された場合、 --diff は正確なファイルサイズの差異を報告しません。

AccessKey ペアを使用した OSS へのアクセス

デフォルトでは、EMR クラスターは明示的な認証情報を必要とせずに OSS にアクセスします。EMR クラスターの外部で Jindo DistCp を実行する場合、または AccessKey を使用しないアクセスがサポートされていない場合は、 --ossKey、 --ossSecret、および --ossEndPoint を使用して認証情報を指定します：

jindo distcp --src /data/incoming/hourly_table --dest oss://<your_bucket>/hourly_table --ossKey <your-access-key-id> --ossSecret <your-access-key-secret> --ossEndPoint oss-cn-hangzhou.aliyuncs.com --parallelism 20

OSS アーカイブストレージまたは低頻度アクセスストレージへの書き込み

--policy を使用して、コピー中にデータを低コストの OSS ストレージクラスに直接書き込みます：

archive：アーカイブストレージクラス

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

ia：低頻度アクセス (IA) ストレージクラス

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

未完了のアップロードのクリーンアップ

コピージョブが中断されると、部分的にアップロードされたファイルがコピー先バケットに残ることがあります。OSS はこれらをアップロード ID で追跡しますが、通常のディレクトリ一覧には表示されません。

--cleanUpPending を追加すると、ジョブの終了時にこれらの未完了のアップロードが自動的に削除されます：

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

または、OSS コンソールで手動でクリーンアップすることもできます。

Amazon S3 から OSS へのコピー

S3 の認証情報とエンドポイントを --s3Key、 --s3Secret、および --s3EndPoint で指定します：

jindo distcp --src s3a://yourbucket/ --dest oss://<your_bucket>/hourly_table --s3Key <your-s3-key> --s3Secret <your-s3-secret> --s3EndPoint s3-us-west-1.amazonaws.com

コマンドごとに認証情報を渡すのを避けるには、 Hadoop の core-site.xml ファイルに設定します：

<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 --src s3a://smartdata1/ --dest oss://smartdata1/tmp --s3EndPoint s3-us-west-1.amazonaws.com

コピーカウンターの確認

ジョブが完了したら、MapReduce ジョブ出力の DistCp Counters を確認して、転送されたデータ量を確認します：

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

コピー中に圧縮または展開が適用された場合、 Bytes Destination Copied と Bytes Source Read が異なる場合があります。