Hadoop DistCp実践ガイド2020年版

Hadoop DistCp (distributed copy, でぃすとしーぴー、でぃすとこぴー) は、MapReduceを用いてHadoopクラスタ間でデータコピーするためのツールです。保守運用している場合を除き、おそらく2020年においても運用上の選択肢として残っている最後のMapReduceのツールです。この記事では、DistCpの紹介と実践的な使い方の基本について説明していきます。内容としては以下の通りです。

• Distcpの概要と原理
• 実践DistCp
  • DistCpにドライランはない
  • コピーとアップデートの挙動の違いを押さえる
  • スナップショットを取得する
  • ソースと宛先、どちらのクラスタでDistCpを実行するか
  • 異なるメジャーバージョン間でのデータ転送にwebhdfsを使う
  • -p オプションの挙動
  • 2つのコピー戦略: uniformizeとdynamic
  • map数の調整
  • 転送帯域

なんで今更DistCp？

DistCpの使い方についてきちんと書いているドキュメントがなかったので書きました。Hadoopのバイブルである象本さえ、DistCpについては本当に簡単なことしか書いておらず、実際の使い方についてまとめているドキュメントがありませんでした。Clouderaのようなベンダーの場合は Cloudera Manager という素晴らしいツールが持つデータレプリケーション機能に包含されていて、ユーザーはボタン一発でクラスタ間データ転送ができるため、DistCpについて細かい話を知る必要はありません。そこで、素のHadoopを使う人のためのDistCpの記事を書いておくことにしました。

DistCpについての機能一覧などの詳細については公式ドキュメントを参照してください。

Hadoop 第3版

hadoop.apache.org

DistCpの概要

DistCp は、MapReduceを用いてHadoopクラスタ間で高速にデータコピーするためのツールで、Apache Hadoop の標準リリースに含まれています。Apache Hadoopは、分散ストレージのHDFS(Hadoop Distributed File System、Hadoop分散ファイルシステム)と、分散コンピューティングフレームワークのYARNから構成されている分散処理フレームワークで、MapReduceはYARN上で動く代表的なアプリケーションの一つです。 Hadoopクラスタ間と書きましたが、正確には分散ストレージ間と言った方が正しいでしょう。DistCpは、HDFSだけでなく、Amazon S3 やAzure Storage といったオブジェクトストレージにも対応しています。

DistCpはコマンドラインツールで、以下のような形式で実行します。

$ hadoop distcp hdfs://cluster1/foo/bar hdfs://cluster2/foo

これは、cluster1というHDFSクラスタの、 /foo/bar というパスを、cluster2 というHDFSクラスタの、 /foo というディレクトリにコピーする、というコマンドとなります。

DistCpの原理

DistCpは、MapReduceフレームワークで動作します。まず、MapReduceについて簡単におさらいします。MapReduceは、複数のノードで別個に計算処理を行うMap、特定のキーごとにデータを転送して集約するShuffle、集約されたデータに対し、Mapと同様、ノードごとに独立して処理を行うReduceという3つのフェーズで分散処理を行うフレームワークです。
以下の図は、MapReduceの処理の流れを表しています。

DistCpは、Map処理のみを使い、何も計算せず(恒等関数)、入力と出力を別のクラスタで行うという形でMapReduceを使用しています。

以下の図は、DistCpの処理の流れを表しています。

DistCpのソース(読み込み元)と宛先(書き込み先)はURIで表されます。先程の例では、宛先を hdfs://cluster2/foo としましたが、この宛先は s3a://bucket1/foo でも問題なく動作します。これは、S3上の bucket1 というバケットの配下にある foo という名前空間にデータをコピーすることを意味します。

実践DistCp: ドライランはない

DistCpは、非常に大規模かつ不可逆変更を行うツールであるにも関わらず、ドライランに相当する機能が存在しないという点に注意してください。ドライランがないということは、十分に検証クラスタでテストした後、本番での実行が成功することを、神(あるいはあなたが信仰する何か)に祈るしかなくなります。そして大抵の場合その祈りが届くことはありません。頑張りましょう。

ドライランについては6年間オープンしているJIRAがありますので、我こそはという方は実装お待ちしています。

issues.apache.org

実践DistCp: コピーとアップデートの挙動の違いを押さえる

hadoop distcp コマンドは、何もオプションをつけない場合は、コピーという挙動になります。これは、以下の操作を行います。

• ソースにパスが存在し、宛先に存在しない場合はコピーする
• ソースと宛先に同じパスが存在する場合は何もしない
• ソースにパスが存在せず、宛先に存在する場合は何もしない

hadoop distcp -update では、以下のように挙動が変わります。

• ソースにパスが存在し、宛先に存在しない場合はコピーする
• ソースと宛先に同じパスが存在する場合、チェックサムなどコンテンツの中身を確認し、コンテンツが異なる場合はコピーする。コンテンツが同一の場合は何もしない
• ソースにパスが存在せず、宛先に存在する場合は何もしない

hadoop distcp -update -delete では、以下のように挙動が変わります。

• ソースにパスが存在し、宛先に存在しない場合はコピーする
• ソースと宛先に同じパスが存在する場合、チェックサムなどコンテンツの中身を確認し、コンテンツが異なる場合はコピーする。コンテンツが同一の場合は何もしない
• ソースにパスが存在せず、宛先に存在する場合はそのパスを削除する

これらの挙動をまとめると、以下の図のようになります。

hadoop distcp に -update をつける場合、コンテンツの中身を比較するため、オーバーヘッドが発生します。そのため、-updateなしに比べて処理性能が落ちることに注意してください。

DistCpのコピーとアップデートの挙動の違いは間違えやすく、そしてその間違いが重大な事故を起こしてしまう可能性がありますので絶対に覚えてください。

以下の2つの例を見てください。

# 例1
$ hadoop distcp hdfs://cluster1/foo/bar hdfs://cluster2/foo
# 例2: 誤った方法
$ hadoop distcp -update hdfs://cluster1/foo/bar hdfs://cluster2/foo

例1は、cluster2/foo の直下に cluster1/foo/bar をコピーするので、結果として cluster2/foo/bar が作成されます。
例2は、 cluster2/foo を cluster1/foo/bar の内容でアップデートするので、 cluster2/foo/bar は作成されず、cluster2/foo のコンテンツが cluster1/foo/bar と同じものになります。

図にすると以下のようになります。

この一例だけだとピンとこないかもしれませんので、もっと実務上実行する可能性のあるコマンドでみてみましょう。

# 例3
$ hadoop distcp hdfs://cluster1/user/sato hdfs://cluster2/user
# 例4: 誤った方法
$ hadoop distcp -update -delete hdfs://cluster1/user/sato hdfs://cluster2/user

hadoop distcp の -delete オプションは -update オプションと一緒に使わないと利用できないオプションで、ソースクラスタには存在しないけど宛先クラスタには存在する全てのパスを削除します。つまり、-delete を付与すると、ソースと宛先のコンテンツが全く同一のものとなります。
例3は、 cluster1/user/sato を、 cluster2/user/ にコピーします。よって、cluster2/user/sato が作成されます。
例4は、 cluster2/user のコンテンツが、cluster1/user/sato と全く同じものになります。つまり、 /user ディレクトリ配下に存在する全てのユーザーデータが完全に削除され、その代わりにユーザ sato のコンテンツだけが置かれるようになります。

図に表すと、以下のようになります。

「ゴミ箱機能があるから即座に削除されることはないのでは？」と思うかもしれませんが、DistCpのバグでゴミ箱は機能しません。この問題は2020/07/15現在未解決です。詳細については以下のJIRAも参照してください。

issues.apache.org

運用者はこのコマンドを誤って実行した時点で、即座に緊急事態のアラートを出さなければいけなくなるでしょう。

この例4は、正しくは以下のように書くべきでした。

# 例4: 誤った方法
$ hadoop distcp -update -delete hdfs://cluster1/user/sato hdfs://cluster2/user
# 例5: 例4の正しい書き方
$ hadoop distcp -update -delete hdfs://cluster1/user/sato hdfs://cluster2/user/sato

では、ここでもう一つの例を紹介しましょう。cluster2に既に/userが存在するときに、以下のコマンドを実行すると何が起きるでしょうか。

# 例6: 誤った方法
$ hadoop distcp hdfs://cluster1/user hdfs://cluster2/user

これが、-update ( -delete ) がついていたならば、問題なかったかもしれません。しかし、今回は -update がついていません。よって、 cluster1/user が cluster2/user の配下にコピーされます。つまり、 cluster2/user/user が作成されます。これは、多くの運用者にとって意図した挙動ではないでしょう。

このとき、安易に cluster2/user/user を削除することはできません。なぜなら、 cluster2/user/user というディレクトリはコピー前から存在していた可能性があり、その中にコンテンツが存在していた可能性があるからです。一度混じってしまえば、cluster1由来のコンテンツとcluster2オリジナルのコンテンツをふるい分けるのは困難でしょう。-update オプションがないときも決して油断してはいけません。
cluster1の/userをcluster2の/userにコピーする場合、以下のように書くべきでした。

# 例6: 誤った方法
$ hadoop distcp hdfs://cluster1/user hdfs://cluster2/user
# 例7: 例6の正しい書き方
$ hadoop distcp hdfs://cluster1/user hdfs://cluster2/

図に表すと、以下のようになります。

手動・自動での実行に関わらず、パスの確認は絶対に最後の最後まで確実に行うようにしてください。

実践DistCp: スナップショットを取得する

DistCpは、通常非常に膨大な時間がかかります。クラスタ全体のデータ転送の場合、1日や2日は当たり前で、1週間や1ヶ月に渡って転送し続ける、ということは頻繁に起こります。DistCpはMapReduce実行前に対象パスの一覧を取得しますので、転送中にソースファイルが変化しても一切考慮することはできません。大抵の場合、転送中にファイルが削除され、何日もかけたDistCpが失敗することになるでしょう。運良く転送に成功したとしても、コンテンツの中身に不整合が発生していれば、Hive等の別のアプリケーションでの処理結果が意図しないものとなり、いいことは一つもありません。そのため、ソースはスナップショットを指定するのが鉄則です。
スナップショットの取得は、以下の2つのコマンドを順番に実行します。

$ hdfs dfsadmin -allowSnapshot hdfs://cluster1/foo/bar
$ hdfs dfs -createSnapshot hdfs://cluster1/foo/bar snapshot1

hdfs dfsadmin -allowSnapshot は hdfs ユーザでないと実行できませんが、hdfs dfs -createSnapshot は、対象ディレクトリの権限を持っている一般ユーザでも実行可能です。上記コマンドを実行すると、 hdfs://cluster1/foo/bar/.snapshot/snapshot1 というディレクトリが作成され、この配下には hdfs://cluster1/foo/bar のコンテンツと全く同じハードリンクが作成されます。
snapshot1はスナップショット名なので、自由に変更してコマンドを実行してください。

スナップショットを使ったDistCpは以下のように記述します。

$ hadoop distcp hdfs://cluster1/foo/bar/.snapshot/snapshot1 hdfs://cluster2/foo

実践DistCp: ソースと宛先、どちらのクラスタでDistCpを実行するか

DistCpは、基本的には宛先クラスタ側で実行することを推奨します。DistCpを宛先クラスタ側で実行しなければならないケースとしては以下のようなものがあります。

• 非セキュアクラスタからセキュアクラスタにデータをコピーする場合
• 低いメジャーバージョンのクラスタから高いメジャーバージョンにデータをコピーする場合

また、新規クラスタへのデータ移行の場合、ソースクラスタは通常業務のアプリケーションが稼働している一方、宛先クラスタは大抵の場合本番稼働前なので、ソースクラスタの負荷を増やさずに、宛先のリソースを有効活用することができます。

DistCpをソースクラスタで実施しなければいけないケースもあります。例えば、セキュアクラスタから非セキュアクラスタへデータを転送する場合です。

Clouderaの以下のドキュメントの記載を引用します。

docs.cloudera.com

You can use DistCp and WebHDFS to copy data between a secure cluster and an insecure cluster. Note that when doing this, the distcp commands should be run from the secure cluster.

セキュアクラスタにおけるDistCpの方法についてはこの記事では扱いませんが、DistCpをどちらのクラスタで実施するかを検討する場合には頭の片隅にとどめておいてください。

実践DistCp: 異なるメジャーバージョン間でのデータ転送にwebhdfsを使う

webhdfsプロトコルを使うことで、メジャーバージョンの低いバージョンから高いバージョンへのデータ転送を行うことができます。

$ hadoop distcp webhdfs://cluster1/foo/bar hdfs://cluster2/foo

以下は参考リンクです。

docs.cloudera.com

実践DistCp: -p オプションの挙動

デフォルトでは、DistCpはファイル属性等はコピーしません。ファイル属性をコピーするには -p オプションを使いますが、このオプションの挙動には様々な制約事項が存在します。例えば、 -update オプションはコンテンツの中身が同一のパスに対してはコピーを実施しませんが、このときファイル属性だけが違っていてもその属性を更新したりはしません。
以下の例で、両クラスタに /foo/bar/file1 というファイルがあるとします。

$ hadoop distcp -update hdfs://cluster1/foo/bar hdfs://cluster2/foo

このとき、cluster1/foo/bar/file1 のパーミッションが644で、 cluster2/foo/bar/file1 のパーミッションが600となっていて、ファイルのコンテンツが全く同一である場合、cluster2/foo/bar/file1 のパーミッションは 600 のまま変更されません。

別の例を紹介しましょう。 -pt オプションを使うと更新日時などを保持できますが、このオプションは、NameNodeの設定の一つ、 dfs.namenode.accesstime.precision (デフォルト1時間) が0(無効)の場合利用できません。dfs.namenode.accesstime.precision を 0 にしたまま以下のコマンドを実行しても、失敗します。

$ hadoop distcp -pt hdfs://cluster1/foo/bar hdfs://cluster2/foo

このとき、以下のようなエラーが出力されます。

Error: org.apache.hadoop.ipc.RemoteException(java.io.IOException): Access time for hdfs is not configured. Please set dfs.namenode.accesstime.precision configuration parameter.

アクセス時間の設定はパフォーマンスの最適化のために0にするのが推奨で、Ambari / HDP はデフォルト0になっていますが、コミュニティ版もClouderaもデフォルト1時間なので、設定そのものを知らない人も多いと思います。 -p オプションを使うときはドキュメントを読むだけで設計せず、必ず検証をしてください。

実践DistCp: 2つのコピー戦略: uniformizeとdynamic

DistCpが各Mapタスクに処理対象のパスを振り分ける戦略は2タイプ存在します。デフォルトはuniformizeという、データサイズで分割する方法です。例えば転送対象のデータが100TBあり、mapタスクを1000で設定した場合、各mapタスクは100GBのデータを転送するように、ファイルパスを振り分けられます。この挙動は、ソースコードを読めばわかりますが、リストされたファイルを上から順に取り出していきサイズを足していき、転送対象の全データサイズ/map数を超えたら次のmapタスクに渡す、という操作を行っています。

github.com

理想的なHDFSの環境ではこれで問題ないのですが、小さいファイルが大量にある環境の場合は、uniformizeではうまくいきません。
uniformizeでは、どれだけたくさんのファイルがあっても、一定のサイズを超えない限りはそれらのファイルが1mapタスクに割り当てられてしまいます。割り当てられるファイルは、ファイルリストの上から順にファイルを取り出されます。ファイルリストは、単純に対象ディレクトリの配下のファイル・ディレクトリを再帰的にリストしているだけなので、同一ディレクトリのファイルは一箇所に固まっています。その結果、あるディレクトリのファイルは1タスクに集中することになります。
1ファイルに対するHDFSアクセスは非常に遅いです。環境にもよりますが、1ファイルあたり数msのオーダーは見たほうがいいでしょう。そのため、スモールファイルが多いストレージでは、データ転送速度は非常に遅くなります。そして、多くの場合、スモールファイルは局所化しています。これはすなわち、特定のディレクトリにスモールファイルが集中していることを意味します。
まとめると、特定のディレクトリに集中したスモールファイル群がまとめて1つのmapタスクに割り当てられる結果、mapタスクのスキューが発生し、そのmapタスクだけが極端に遅くなるという現象が発生します。

このような環境では、dynamic というもう一つのコピー戦略を使います。dynamic はファイル数でタスクあたりの割当を分割するオプションです。例えば、1億ファイルあるシステムで1000mapタスクで処理を分割する場合、1タスクあたり10万ファイルを担当することになります。
dynamicオプションを使う場合、uniformizeと逆に、極端にファイルサイズが大きいデータが集中しているケースに注意してください。ファイルサイズを考慮しないでデータを分割するため、特定のタスクだけ極端に大きなデータを処理しなければいけないというリスクが発生します。転送対象のデータ特性は必ず事前に調査しましょう。
dynamic 戦略を使うには、以下のようにオプションを与えます。

$ hadoop distcp -strategy dynamic hdfs://cluster1/foo/bar hdfs://cluster2/foo

実践DistCp: map数の調整

デフォルトではDistCpは20mapタスクしか使用しません。データ量やリソース状況に応じて、map数の調整をしたほうがいいでしょう。以下の例は、map数を100とする場合の例です。

$ hadoop distcp -m 100 hdfs://cluster1/foo/bar hdfs://cluster2/foo

map数の調整は、基本的なHadoopアプリケーションと同様、ストレージIOやリソースに応じて調整する必要があります。リソースをフルに使えるのであれば、総ディスク数の1～2倍くらいにしておくのがいいと思いますが、例えばスモールファイル中心のクラスタの場合IOよりもCPU依存になるはずなので、CPUコア数からタスク数を計算した方がいいかもしれませんし、クラスタのリソースが逼迫している状態であればむしろmap数を減らしてゆっくり処理した方がいいかもしれません。このあたりの計算に自信がなければ、まずはデフォルトで試験的に転送してみて、転送速度を計算した上で必要があればチューニングするという程度でいいと思います。

実践DistCp: 転送帯域

ネットワークの帯域リソースが逼迫している場合は、転送用の帯域を制御した方がいいでしょう。以下のように設定することで、1mapあたりの転送帯域を10MB/sに抑えることができます。

$ hadoop distcp -bandwidth 10 hdfs://cluster1/foo/bar hdfs://cluster2/foo

この記事で書いていないこと

ただコピーするといっても、細かい要件はプロジェクトによって異なり、それに応じてDistCpの様々な機能を活用していく必要があります。
この記事でカバーしていない内容は以下の通りです。

• snapshot diff を使った定常的な差分バックアップ
• HAクラスタでのDistCp
• セキュアクラスタでのDistCp
• オブジェクトストレージを対象としたDistCp

そして、クラスタ移行という話になったときは、必要な作業はDistCpだけではありません。例えば、HiveメタストアDBのデータ移行や、管理ツールのデータ移行など、考えるべき課題は他にもあります。これらについて、最新の情報をベースに体系的にまとめられた書籍は存在しないので、もし自信がないという場合はCloudera等のベンダーに相談することをおすすめします。

参考リンク

既出も含めて、参考リンクをまとめておきます。

• Apache Hadoop の DistCp のドキュメント
• Cloudera Enterprise のドキュメント
• HDP のドキュメント

謝辞

本記事の執筆にあたり、以下の方々にレビューしていただきました。この場を借りてお礼申し上げます。(順不同、敬称略)

• chezou
• d1ce_
• kernel023
• kmizumar
• moriyoshit
• nyaxt
• takabow