DistCpガイド

基本的な使用方法

DistCpの最も一般的な呼び出しは、クラスタ間コピーです。

bash$ hadoop distcp hdfs://nn1:8020/foo/bar \
hdfs://nn2:8020/bar/foo

これにより、nn1上の/foo/barの下のネームスペースが一時ファイルに展開され、その内容がマップタスクのセットに分割され、各NodeManager上でnn1からnn2へのコピーが開始されます。

コマンドラインで複数のソースディレクトリを指定することもできます。

bash$ hadoop distcp hdfs://nn1:8020/foo/a \
hdfs://nn1:8020/foo/b \
hdfs://nn2:8020/bar/foo

または、同等に、-fオプションを使用してファイルから指定することもできます。

bash$ hadoop distcp -f hdfs://nn1:8020/srclist \
hdfs://nn2:8020/bar/foo

ここで、srclistには以下が含まれています。

hdfs://nn1:8020/foo/a
hdfs://nn1:8020/foo/b

複数のソースからコピーする場合、2つのソースが衝突すると、DistCpはエラーメッセージとともにコピーを中止しますが、宛先での衝突は、指定されたオプションに従って解決されます。デフォルトでは、宛先に既に存在するファイルはスキップされます（つまり、ソースファイルで置き換えられません）。各ジョブの最後にスキップされたファイル数が報告されますが、一部のファイルでコピアが失敗したが、後の試行で成功した場合、不正確になる可能性があります。

各NodeManagerがソースファイルシステムと宛先ファイルシステムの両方にアクセスして通信できることが重要です。HDFSの場合、ソースと宛先の両方で同じバージョンのプロトコルを実行するか、下位互換性のあるプロトコルを使用する必要があります。[バージョンの間のコピー](#Copying_Between_Versions_of_HDFS)を参照してください。

コピー後、ソースと宛先のリストを生成して相互にチェックし、コピーが実際に成功したことを確認することをお勧めします。DistCpはMap/ReduceとファイルシステムAPIの両方を使用するため、3つのいずれかまたはそれらの間の問題が、コピーに悪影響を与え、サイレントに影響を与える可能性があります。-updateを有効にして2回目のパスを実行することに成功した人もいますが、ユーザーは試行する前にその意味を理解する必要があります。

別のクライアントがまだソースファイルに書き込んでいる場合、コピーは失敗する可能性があることにも注意してください。宛先に書き込まれているファイルを上書きしようとすると、HDFSでも失敗する可能性があります。ソースファイルがコピーされる前に（再）移動されると、FileNotFoundExceptionでコピーが失敗します。

DistCpで使用可能なすべてのオプションの詳細については、コマンドラインリファレンスを参照してください。

更新と上書き

-updateは、ターゲットに存在しないか、ターゲットバージョンと異なるソースからのファイルをコピーするために使用されます。-overwriteは、ターゲットに存在するターゲットファイルを上書きします。

更新と上書きオプションは、ソースパスの処理方法がデフォルトとは微妙に異なるため、特別な注意が必要です。/source/first/と/source/second/から/target/へのコピーを検討してください。ソースパスには次の内容が含まれています。

hdfs://nn1:8020/source/first/1
hdfs://nn1:8020/source/first/2
hdfs://nn1:8020/source/second/10
hdfs://nn1:8020/source/second/20

-updateまたは-overwriteを指定せずにDistCpを呼び出すと、DistCpのデフォルトでは、/targetの下にfirst/とsecond/ディレクトリが作成されます。したがって

distcp hdfs://nn1:8020/source/first hdfs://nn1:8020/source/second hdfs://nn2:8020/target

は、/targetに次の内容を作成します。

hdfs://nn2:8020/target/first/1
hdfs://nn2:8020/target/first/2
hdfs://nn2:8020/target/second/10
hdfs://nn2:8020/target/second/20

-updateまたは-overwriteを指定すると、ソースディレクトリの**内容**がターゲットにコピーされ、ソースディレクトリ自体はコピーされません。したがって

distcp -update hdfs://nn1:8020/source/first hdfs://nn1:8020/source/second hdfs://nn2:8020/target

は、/targetに次の内容を作成します。

hdfs://nn2:8020/target/1
hdfs://nn2:8020/target/2
hdfs://nn2:8020/target/10
hdfs://nn2:8020/target/20

拡張として、両方のソースフォルダに同じ名前のファイル（たとえば、0）が含まれている場合、両方のソースは宛先の/target/0にエントリをマップします。この競合を許可するのではなく、DistCpは中止します。

次に、次のコピー操作を考えてみましょう。

distcp hdfs://nn1:8020/source/first hdfs://nn1:8020/source/second hdfs://nn2:8020/target

ソース/サイズ：

hdfs://nn1:8020/source/first/1 32
hdfs://nn1:8020/source/first/2 32
hdfs://nn1:8020/source/second/10 64
hdfs://nn1:8020/source/second/20 32

宛先/サイズ：

hdfs://nn2:8020/target/1 32
hdfs://nn2:8020/target/10 32
hdfs://nn2:8020/target/20 64

は次のようになります。

hdfs://nn2:8020/target/1 32
hdfs://nn2:8020/target/2 32
hdfs://nn2:8020/target/10 64
hdfs://nn2:8020/target/20 32

ファイルの長さと内容が一致するため、1はスキップされます。2はターゲットに存在しないためコピーされます。内容がソースと一致しないため、10と20は上書きされます。

-updateを使用する場合、ファイルの長さと内容が一致するため、1はスキップされます。2はターゲットに存在しないためコピーされます。内容がソースと一致しないため、10と20は上書きされます。ただし、-appendも追加で使用されると、10のみが上書きされ（ソースの長さが宛先より短い）、20にはファイルの変更が追加されます（ファイルが宛先の元の長さまで一致する場合）。

-overwriteを使用すると、1も上書きされます。

同期

-diffオプションは、スナップショット差分を使用して、ソースクラスタからターゲットクラスタにファイルを同期します。スナップショット差分リスト内のファイルをコピー、名前変更、削除します。

-diffオプションを使用する場合は、-updateオプションを含める必要があります。

現時点では、ほとんどのクラウドプロバイダーは同期と適切に連携しません。

使用方法

hadoop distcp -update -diff <from_snapshot> <to_snapshot> <source> <destination>

例

hadoop distcp -update -diff snap1 snap2 /src/ /dst/

上記のコマンドは、/src/のsnapshot snap1からsnap2（つまり、snap1からsnap2のスナップショット差分）の変更を/dst/に適用します。明らかに、/src/にはsnap1とsnap2の両方のスナップショットが必要です。しかし、宛先/dst/にも、この場合はsnap1と同様に<from_snapshot>と同じ名前のスナップショットが必要です。宛先/dst/には、snap1以降に新しいファイル操作（作成、名前変更、削除）があってはなりません。このコマンドが完了すると、宛先/dst/に新しいスナップショットsnap2は**作成されません**。

-diffオプションを使用するには、-updateが必要です。

たとえば、/src/で、snap1の作成後、snap2の作成前に1.txtが追加され、2.txtが削除された場合、上記のコマンドは/src/から/dst/に1.txtをコピーし、/dst/から2.txtを削除します。

同期の動作は、以下の実験を使用して詳しく説明します。

# Create source and destination directories
hdfs dfs -mkdir /src/ /dst/
# Allow snapshot on source
hdfs dfsadmin -allowSnapshot /src/
# Create a snapshot (empty one)
hdfs dfs -createSnapshot /src/ snap1
# Allow snapshot on destination
hdfs dfsadmin -allowSnapshot /dst/
# Create a from_snapshot with the same name
hdfs dfs -createSnapshot /dst/ snap1

# Put one text file under /src/
echo "This is the 1st text file." > 1.txt
hdfs dfs -put 1.txt /src/
# Create the second snapshot
hdfs dfs -createSnapshot /src/ snap2

# Put another text file under /src/
echo "This is the 2nd text file." > 2.txt
hdfs dfs -put 2.txt /src/
# Create the third snapshot
hdfs dfs -createSnapshot /src/ snap3

hadoop distcp -update -diff snap1 snap2 /src/ /dst/

hdfs dfs -rm -skipTrash /dst/1.txt

hadoop distcp -update -diff snap1 snap3 /src/ /dst/

hdfs dfs -rm -skipTrash /dst/2.txt
# Create snap2 at destination, it contains 1.txt
hdfs dfs -createSnapshot /dst/ snap2

# Delete 1.txt from source
hdfs dfs -rm -skipTrash /src/1.txt
# Create snap4 at source, it only contains 2.txt
hdfs dfs -createSnapshot /src/ snap4

hadoop distcp -update -diff snap2 snap4 /src/ /dst/

raw名前空間拡張属性の保持

このセクションはHDFSのみに適用されます。

ターゲットとすべてのソースパス名が/.reserved/raw階層内にある場合、「raw」名前空間拡張属性は保持されます。「raw」xattrは、暗号化メタデータなどの内部関数でシステムによって使用されます。それらは、/.reserved/raw階層を介してアクセスした場合にのみユーザーに表示されます。

raw xattrは、/.reserved/rawプレフィックスが提供されているかどうかに基づいてのみ保持されます。-p（保持、下記参照）フラグは、raw xattrの保持に影響しません。

raw xattrの保持を防ぐには、ソースパスとターゲットパスのいずれにも/.reserved/rawプレフィックスを使用しないようにしてください。

ソースパスとターゲットパスのサブセットのみに/.reserved/rawプレフィックスが指定されている場合、エラーが表示され、0以外の終了コードが返されます。

DistCpドライバ

DistCpドライバコンポーネントは、次の役割を担います。

• コマンドラインでDistCpコマンドに渡された引数を、

  • OptionsParserを介して、および
  • DistCpOptionsSwitchを介して解析します。

• コマンド引数を適切なDistCpOptionsオブジェクトに組み立て、DistCpを初期化します。これらの引数には、以下が含まれます。

  • ソースパス
  • ターゲット場所
  • コピーオプション（更新コピー、上書き、保持するファイル属性など）

• コピー操作のオーケストレーション

  • コピー対象ファイル一覧を生成するためのコピー一覧ジェネレータの呼び出し。
  • コピーを実行するためのHadoop Map-Reduceジョブの設定と起動。
  • オプションに基づき、Hadoop MRジョブへのハンドルをすぐに返すか、完了まで待機するかを選択。

パーサー要素は、コマンドラインからのみ実行されます（またはDistCp::run()が呼び出された場合）。DistCpクラスは、DistCpOptionsオブジェクトを構築し、DistCpオブジェクトを適切に初期化することで、プログラム的にも使用できます。

コピー一覧ジェネレータ

コピー一覧ジェネレータクラスは、ソースからコピーするファイル/ディレクトリのリストを作成する役割を担います。ソースパス（ワイルドカードを含むファイル/ディレクトリ）の内容を調べ、DistCp Hadoopジョブで使用するために、コピーが必要なすべてのパスをSequenceFileに記録します。このモジュールにおける主なクラスには以下が含まれます。

1. CopyListing：コピー一覧ジェネレータ実装によって実装されるべきインターフェース。具体的なCopyListing実装を選択するためのファクトリメソッドも提供します。
2. SimpleCopyListing：複数のソースパス（ファイル/ディレクトリ）を受け入れ、それぞれの下にある個々のファイルとディレクトリを再帰的にリストしてコピーするためのCopyListingの実装。
3. GlobbedCopyListing：ソースパス内のワイルドカードを展開するCopyListingの別の実装。
4. FileBasedCopyListing：指定されたファイルからソースパスリストを読み取るCopyListingの実装。

DistCpOptionsでソースファイルリストが指定されているかどうかに基づいて、ソースリストは次のいずれかの方法で生成されます。

1. ソースファイルリストがない場合、GlobbedCopyListingが使用されます。すべてのワイルドカードが展開され、すべての展開がSimpleCopyListingに転送され、SimpleCopyListingは順番に（各パスの再帰的下降によって）リストを構築します。
2. ソースファイルリストが指定されている場合、FileBasedCopyListingが使用されます。ソースパスは指定されたファイルから読み取られ、GlobbedCopyListingに転送されます。その後、上記のようにリストが構築されます。

CopyListingインターフェースのカスタム実装を提供することで、コピーリストの構築方法をカスタマイズできます。パスのコピーの考慮方法において、DistCpの動作は従来のDistCpとは異なります。

コピーすべきではないファイルのフィルタリングをカスタマイズするには、サポートされているCopyFilterインターフェースの現在の実装を渡すか、新しい実装を作成できます。これは、DistCpOptionsでdistcp.filters.classを設定することで指定できます。

1. distcp.filters.classを"RegexCopyFilter"に設定します。この実装を使用する場合は、フィルタリングに使用される正規表現を含む"CopyFilter" distcp.filters.fileも渡す必要があります。java.util.regex.Patternで指定された正規表現をサポートします。
2. distcp.filters.classを"RegexpInConfigurationFilter"に設定します。"DistCpOptions"でdistcp.exclude-file-regexパラメータも使用して正規表現を渡す必要があります。java.util.regex.Patternで指定された正規表現をサポートします。"RegexCopyFilter"と比較して、より動的なアプローチです。
3. distcp.filters.classを"TrueCopyFilter"に設定します。上記いずれのオプションも指定されていない場合、デフォルトの実装として使用されます。

従来の実装では、ターゲットに確実にコピーする必要があるパスのみがリストされます。たとえば、ファイルが既にターゲットに存在する場合（そして-overwriteが指定されていない場合）、そのファイルはMapReduceコピージョブでは考慮されません。設定時に（つまり、MapReduceジョブの前に）これを決定するには、ファイルサイズとチェックサムの比較が必要であり、時間がかかる可能性があります。

新しいDistCpは、MapReduceジョブまでこのようなチェックを延期するため、設定時間が短縮されます。これらのチェックが複数のマップに並列化されるため、パフォーマンスがさらに向上します。

InputFormatsとMapReduceコンポーネント

InputFormatsとMapReduceコンポーネントは、ファイルとディレクトリのソースパスから宛先パスへの実際のコピーを担当します。コピー一覧生成中に作成された一覧ファイルは、コピーの実行時にこの時点で消費されます。ここで重要なクラスには以下が含まれます。

• UniformSizeInputFormat：org.apache.hadoop.mapreduce.InputFormatの実装により、マップ間の負荷分散において従来のDistCpと同等の機能を提供します。UniformSizeInputFormatの目的は、各マップがほぼ同じバイト数のファイルをコピーすることです。適切に、一覧ファイルはパスグループに分割され、各InputSplit内のファイルサイズの合計が他のすべてのマップとほぼ等しくなります。分割が常に完璧というわけではありませんが、その単純な実装により、設定時間が短縮されます。

• DynamicInputFormatとDynamicRecordReader：DynamicInputFormatはorg.apache.hadoop.mapreduce.InputFormatを実装し、DistCpの新機能です。一覧ファイルはいくつかの「チャンクファイル」に分割され、チャンクファイルの正確な数は、Hadoopジョブで要求されたマップ数の倍数になります。各マップタスクは、ジョブが起動される前に、チャンクの1つを（チャンクをタスクのIDに名前変更することで）「割り当て」られます。パスはDynamicRecordReaderを使用して各チャンクから読み取られ、CopyMapperで処理されます。チャンク内のすべてのパスが処理されると、現在のチャンクは削除され、新しいチャンクが取得されます。チャンクがなくなるまで、このプロセスが継続されます。この「動的」なアプローチにより、高速なマップタスクは遅いタスクよりも多くのパスを消費できるため、DistCpジョブ全体の速度が向上します。

• CopyMapper：このクラスは、物理的なファイルコピーを実装します。入力パスは、（ジョブの設定で指定された）入力オプションに対してチェックされ、ファイルをコピーする必要があるかどうかが決定されます。ファイルは、次のいずれかがtrueの場合にのみコピーされます。

  • 同じ名前のファイルがターゲットに存在しない。
  • 同じ名前のファイルがターゲットに存在するが、ファイルサイズが異なる。
  • 同じ名前のファイルがターゲットに存在するが、チェックサムが異なり、-skipcrccheckが指定されていない。
  • 同じ名前のファイルがターゲットに存在するが、-overwriteが指定されている。
  • 同じ名前のファイルがターゲットに存在するが、ブロックサイズが異なり（そしてブロックサイズを保持する必要がある）。

• CopyCommitter：このクラスは、DistCpジョブのコミットフェーズを担当し、以下を含みます。

  • ディレクトリパーミッションの保持（オプションで指定されている場合）
  • 一時ファイル、作業ディレクトリなどのクリーンアップ。

マップのサイズ設定

デフォルトでは、DistCpは各マップがほぼ同じバイト数をコピーするように、各マップのサイズを同等にする試みを行います。ファイルは最も細かい粒度であるため、同時コピアー（つまりマップ）の数を増やしても、同時コピーの数や全体のスループットが常に増加するとは限りません。

新しいDistCpは、高速なデータノードが遅いノードよりも多くのバイトをコピーできるように、マップサイズを「動的に」調整する戦略も提供します。-strategy dynamic（アーキテクチャで説明）を使用すると、各マップタスクに固定されたソースファイルセットを割り当てるのではなく、ファイルはいくつかのセットに分割されます。セットの数は、通常、マップ数の2〜3倍になります。各マップは、チャンクにリストされているすべてのファイルを取得してコピーします。チャンクを使い果たすと、新しいチャンクが取得され処理され、チャンクがなくなるまで続きます。

ソースパスを固定されたマップに割り当てないことで、高速なマップタスク（つまりデータノード）はより多くのチャンクを消費し、したがって遅いノードよりも多くのデータをコピーできます。この分散は均一ではありませんが、各マッパーの容量に関して公平です。

動的戦略は、DynamicInputFormatによって実装されます。ほとんどの状況で優れたパフォーマンスを提供します。

長時間実行されるジョブや定期的に実行されるジョブの場合、ソースクラスタと宛先クラスタのサイズ、コピーのサイズ、利用可能な帯域幅に合わせてマップ数を調整することをお勧めします。

HDFSのバージョンの間でのコピー

Hadoopの異なるメジャーバージョン間（例：1.Xと2.Xの間）のコピーを行うには、通常、WebHdfsFileSystemを使用します。以前のHftpFileSystemとは異なり、webhdfsは読み取りと書き込みの両方の操作で使用できるため、DistCpはソースクラスタと宛先クラスタの両方で実行できます。リモートクラスタはwebhdfs://<namenode_hostname>:<http_port>として指定します。Hadoopクラスタの同じメジャーバージョン間（例：2.Xと2.Xの間）のコピーを行う場合は、パフォーマンス向上のためにhdfsプロトコルを使用します。

distcpを使用したワイヤ経由のセキュアコピー

webhdfsがSSLで保護されている場合は、「swebhdfs://」スキームを使用します。詳細については、SWebHDFSのSSL設定を参照してください。

MapReduceおよびその他の副作用

前述のように、マップが入力データの1つのコピーに失敗した場合、いくつかの副作用が発生します。

• -overwriteが指定されていない限り、再実行時に前のマップによって正常にコピーされたファイルは「スキップ済み」としてマークされます。
• マップがmapreduce.map.maxattempts回失敗した場合、残りのマップタスクはキルされます（-iが設定されていない限り）。
• mapreduce.map.speculativeがfinalおよびtrueに設定されている場合、コピーの結果は未定義です。

DistCpとオブジェクトストア

DistCpは、Amazon S3、Azure ABFS、Google GCSなどのオブジェクトストアと連携します。

前提条件

1. オブジェクトストアの実装を含むJARは、すべての依存関係とともにクラスパス上にあります。
2. JARがバンドルされたファイルシステムクライアントを自動的に登録しない限り、ファイルシステムスキーマを実装するクラスを指定するように設定を変更する必要がある場合があります。ASF独自のオブジェクトストアクライアントはすべて自己登録されます。
3. 関連するオブジェクトストアアクセス資格情報は、クラスタ設定で使用可能であるか、またはその他の方法ですべてのクラスタホストで使用可能である必要があります。

DistCpを使用してデータをアップロードできます。

hadoop distcp -direct hdfs://nn1:8020/datasets/set1 s3a://bucket/datasets/set1

データをダウンロードするには

hadoop distcp s3a://bucket/generated/results hdfs://nn1:8020/results

オブジェクトストア間でデータをコピーするには

hadoop distcp s3a://bucket/generated/results \
  wasb://updates@example.blob.core.windows.net

オブジェクトストア内でデータをコピーするには

hadoop distcp wasb://updates@example.blob.core.windows.net/current \
  wasb://updates@example.blob.core.windows.net/old

そして、変更されたファイルのみをコピーするために-updateを使用します。

hadoop distcp -update -numListstatusThreads 20  \
  s3a://history/2016 \
  hdfs://nn1:8020/history/2016

オブジェクトストアはファイルのリスト作成が遅いので、大きなディレクトリツリーで-update操作を実行する場合は、-numListstatusThreadsオプションを設定することを検討してください（制限は40スレッドです）。

オブジェクトストアでDistCp -updateを使用する場合、一般的に、個々のファイルの最終更新時刻と長さのみが比較され、2つのストア間のチェックサムアルゴリズムが異なる場合は、チェックサムは比較されません。

• 異なるチェックサムアルゴリズムを持つ2つのオブジェクトストア間のdistcp -updateは、ファイルコピーをスキップするかどうかを判断するために、ソースファイルとターゲットファイルの最終更新時刻とファイルサイズを比較します。この動作はプロパティdistcp.update.modification.timeによって制御され、デフォルトでtrueに設定されています。ソースファイルがターゲットファイルよりも最近変更されている場合、コンテンツが変更されたと見なされ、ファイルが更新される必要があります。マシン間にクロックスキューがないことを確認する必要があります。ほとんどのオブジェクトストアにはディレクトリの有効なタイムスタンプがあるという事実は無関係です。ファイルのタイムスタンプのみが比較されます。ただし、クライアントコンピュータのクロックをインフラストラクチャのクロックに近づけることが重要であり、これにより、クライアント/HDFSクラスタとオブジェクトストア間のタイムスタンプが一致するようになります。そうでない場合、変更されたファイルが不足したり、頻繁にコピーされたりする可能性があります。

• distcp.update.modification.timeは、2つのストアのいずれかにチェックサム検証がない場合、つまり2つのストア間のチェックサム比較が互換性のない場合にのみ使用されます。プロパティがtrueに設定されている場合でも、2つのストア間に有効なチェックサム比較がある場合は使用されません。

最終更新時刻チェックをオフにするには、core-site.xmlでこれ を設定します。

<property>
        <name>distcp.update.modification.time</name>
        <value>false</value>
</property>

備考

• -atomicオプションは一時データのリネームを引き起こすため、操作終了時のコミット時間が大幅に増加します。さらに、(オプションで)wasb://以外のオブジェクトストアはディレクトリの原子的なリネームを提供しないため、-atomic操作は実際には期待通りの動作をしません。使用を避けてください。

• -appendオプションはサポートされていません。

• -diffおよびrdiffオプションはサポートされていません。

• -skipCrcフラグの値に関係なく、CRCチェックは実行されません。

• パーミッション、ユーザーおよびグループ情報、属性チェックサム、レプリケーションを保持するための-pオプションはすべて、一般的に無視されます。wasb://コネクタは情報を保持しますが、パーミッションを強制適用しません。

• 一部のオブジェクトストアコネクタ（例：S3Aコネクタ）は、出力のメモリ内バッファリングオプションを提供しています。大規模なファイルをコピーする際にこのようなオプションを使用すると、ヒープオーバーフローやYARNコンテナの終了など、何らかのメモリ不足イベントが発生する可能性があります。これは、クラスタとオブジェクトストア間のネットワーク帯域幅が制限されている場合（リモートオブジェクトストアを使用する場合など）、特に一般的です。このようなオプションを無効化/回避し、ディスクバッファリングに依存することをお勧めします。

• 単一のオブジェクトストア内でのコピー操作は、オブジェクトストアが内部的により効率的なCOPY操作を実装している場合でも、Hadoopクラスタ内で実行されます。

  つまり、次のような操作では

  hadoop distcp s3a://bucket/datasets/set1 s3a://bucket/datasets/set2

  各バイトがHadoopワーカーノードに転送され、その後バケットに戻されます。これは遅くなるだけでなく、料金が発生する可能性もあります。

• -directオプションを使用すると、オブジェクトストアのターゲットパスに直接書き込むことができ、それ以外の場合は発生する可能性のある非常にコストのかかる一時ファイルのリネーム操作を回避できます。