Guarantee Mapreduce/Spark task data consistency through Job Committer

ターゲットストレージシステムへのデータの同時書き込みは、分散タスクの基本的な特性です。ノード、プロセス、スレッドレベルでデータを同時に書き込むことで、クラスターのディスクとネットワーク帯域幅を最大限に活用し、大容量のスループットを実現できます。同時書き込みで解決すべき主な課題の 1 つは、データ整合性をいかに確保するかです。具体的には、以下の課題を解決する必要があります。

1. 分散タスクのデータ書き込みプロセスにおいて、中間データを外部から不可視にする方法。

2. 分散タスクの正常完了後、すべての結果データが同時に外部から可視化されることを保証する方法。

3. 分散タスクの失敗時に、すべての結果データを外部から不可視にし、適切にクリーンアップする方法。

4. 予測実行 (Speculative Execution) が有効な場合、同一タスクを実行する複数のタスクの結果データが最終結果に 1 つだけ残るようにする方法。

さらに、タスク失敗時のリトライやジョブの再起動などのジョブ例外も処理する必要があります。MapReduce では Job Committer を使用して、分散書き込みの一貫性を保証しています。Job Committer のさまざまな実装により、MapReduce タスクによるデータ書き込みのあらゆる異常シナリオにおける一貫性が保証されます。Spark は MapReduce の JobCommitter をサポートしており、Spark ジョブによるデータ書き込みの一貫性も JobCommitter を通じて実現されています。

JobCommitter インターフェイス
MapReduce には V1 と V2 の 2 系統の API インターフェイスがあり、パッケージ名の mapred と mapreduce で区別されています。V1 および V2 バージョンの JobCommitter 抽象インターフェイスは基本的に同じです。ここでは org.apache.hadoop.mapreduce.OutputCommitter を例に、主なインターフェイスの定義を紹介します。

このインターフェイスの呼び出しタイミングと順序から、MapReduce タスクが JobCommitter を通じて動作する仕組みを概観できます。

1. ジョブの初期化時に setupJob を呼び出し、ジョブの作業ディレクトリの設定など、ジョブレベルの初期化作業を行います。

2. 同一のジョブが既に実行中の場合、isCommitJobRepeatable を呼び出して継続の可否を判断します。

3. タスクの初期化時に setupTask を呼び出し、タスクの作業ディレクトリやタスク出力先ディレクトリなどの初期化作業を行います。

4. タスクの出力が既に存在する場合、isRecoverySupport でリカバリーがサポートされているかを判断します。サポートされている場合は recoverTask を呼び出し、タスクの再計算を回避します。

5. タスクの実行に失敗した場合、abortTask を呼び出してタスクの出力をクリーンアップします。

6. タスクの実行に成功した場合、commitTask を呼び出します。

7. すべてのタスクが完了した場合、commitJob を呼び出します。

8. ジョブが失敗した場合、abortJob を呼び出します。

JobCommitter の基本的な仕組みは、分散データベースの二相コミットプロトコルに似た方式に基づいていることがわかります。まずタスクがコミットを行い、タスク内で主な作業が完了します。appmaster (ApplicationMaster) がすべてのタスクのコミット成功情報を受け取った後、ジョブのコミットを実行して最終的なコミット作業を完了します。二相コミットプロトコルによるデータ整合性の実現には、主に 2 つの重要な要件があります。

1. ジョブのコミット前、データは外部から不可視であり、かつロールバック可能であること。

2. ジョブのコミットプロセスはできるだけ短くすることが望ましく、理想的にはアトミック操作であること。コミットプロセスが長いと途中で失敗するリスクが高まり、失敗するとデータが中間状態になり、データ整合性の要件を満たせなくなります。

MapReduce では、FileOutputCommitter が最も一般的な Job Committer 実装です。HDFS へのデータ書き込み時、二相コミットプロトコルの 2 つの要件を完全に満たします。

FileOutputCommitter

以下に、FileOutputCommitter の主なインターフェイスの具体的な実装の詳細を簡潔に紹介します。FileOutputCommitter は主に 4 つのディレクトリを扱います。

最終ディレクトリ：$dest/
ジョブ一時ディレクトリ：$dest/_temporary/$appAttemptId/
タスク一時ディレクトリ：$dest/_temporary/$appAttemptId/_temporary/$taskAttemptID/
タスク出力ディレクトリ：$dest/_temporary/$appAttemptId/$taskAttemptID/
JobCommitter の実行プロセス全体を図に示します。

1. setupJob：ジョブ一時ディレクトリを作成します。

2. setupTask：タスク一時ディレクトリと出力ディレクトリを決定します。

3. commitTask：タスク一時ディレクトリを出力ディレクトリにリネームします。

4. abortTask：タスク一時ディレクトリをクリーンアップします。

5. commitJob：ジョブ一時ディレクトリ内のデータ (すべてのタスクの出力ディレクトリ内のファイルを含む) をジョブの最終ディレクトリにマージします。

6. abortJob：ジョブ一時ディレクトリをクリーンアップします。

上記の FileOutputCommitter の実装から、commitJob の前は mapreduce タスクが書き込んだすべてのデータが一時ディレクトリ内にあり、ジョブの最終ディレクトリを読み取っても一時データは読み取られないことがわかります。ジョブの実行プロセスでエラーが発生した場合、一時ディレクトリ内のファイルがクリーンアップされます。FileOutputCommitter の実行中、各ファイルは 2 回のリネーム操作を経ます。1 回目は commitTask で、タスク内で実行されます。複数ノードで実行されるタスクのリネームは並行して実行できます。2 回目は commitJob で、MapReduce または Spark のジョブドライバーによって実行される単一ポイント操作です。commitJob の実行中、ジョブの一時ディレクトリ内のファイルを最終ディレクトリに移動する必要があるため、タイムウィンドウが発生します。このプロセス中に失敗すると、一部のデータが外部に可視化される可能性があり、このタイムウィンドウはファイル数の増加に伴って拡大します。HDFS などの分散ファイルシステムでは、リネーム操作は NameNode 上の関連メタデータの変更のみを伴う非常に効率的な操作であるため、このタイムウィンドウは非常に小さく、大半のシナリオの要件を満たせます。

S3 や OSS などのパブリッククラウドのオブジェクトストレージシステムでは、Rename 操作が直接サポートされておらず、ファイルシステムレベルの Rename 操作は一般的に Copy + Delete 操作に変換されるため、HDFS と比べてコストが大幅に増加します。commitJob は MapReduce または Spark のジョブドライバー側で実行される単一ポイント操作です。スレッドレベルの同時実行最適化が実装されていますが、S3/OSS への書き込みシナリオでは、commitJob のタイムウィンドウが非常に長くなり、ファイル数が多い場合は分単位、さらには時間単位に達する可能性があり、ジョブのパフォーマンスに深刻な影響を与えます。S3/OSS などのオブジェクトストレージシステムへの書き込みのパフォーマンス問題を解決するため、Hadoop コミュニティは FileOutputCommitter V2 バージョンを導入しました。

FileOutputCommitter V2
FileOutputCommitter V2 バージョンのジョブコミットプロセス全体は以下のとおりです。

img

1. setupJob：ジョブ一時ディレクトリを作成します。

2. setupTask：タスク一時ディレクトリを決定します。

3. commitTask：タスク一時ディレクトリのファイルをジョブの最終ディレクトリにリネームします。

4. abortTask：タスク一時ディレクトリをクリーンアップします。

5. commitJob：リネーム操作は不要です。

6. abortJob：ジョブ一時ディレクトリをクリーンアップします。

V2 バージョンでは、最大の違いはタスク出力ディレクトリが省略されたことであり、commitTask 時にファイルが直接ジョブの最終ディレクトリにリネームされることです。ジョブコミットプロセス全体で、すべてのファイルに対する Rename 操作は 1 回だけで済み、この Rename 操作はクラスターノードのすべてのタスクで並行して実行されるため、ジョブドライバーによる単一ポイント実行のボトルネックが解消されます。

FileOutputCommitter V2 は S3/OSS へのデータ書き込みシナリオでのパフォーマンスを大幅に向上させましたが、タスク出力ディレクトリが省略されたため、データ整合性を保証できなくなりました。ジョブの実行中に一部のファイルが既にジョブの最終ディレクトリに移動されているため、一部のタスクが成功し一部のタスクが失敗した場合、中間ファイルが最終ディレクトリに残ってしまいます。

S3/OSS への書き込みなどのシナリオに対して、Hadoop コミュニティや各業界は多数のソリューションを提案しています。基本的な目標は、データ整合性を確保しつつ Rename 操作を完全に回避することです。以下では主に S3ACommitter と JindoOssCommitter について紹介します。これらはそれぞれ Hadoop コミュニティと Alibaba Cloud EMR チームが S3 および OSS 向けに実装した Job Committer です。主に S3/OSS の Multipart Upload 機能に基づいており、基本的な考え方は同じなので、ここではまとめて紹介します。なお、Databricks の DBIO ベースのソリューションや Netflix の Staging committer ソリューションなどもありますが、紙面の都合上、ここでは詳細な紹介は省略します。

オブジェクトストレージシステムの Multipart Upload
S3/OSS へのファイルアップロードには、PUT Object インターフェイスを使用する方法の他に、Multipart Upload というもう 1 つのアップロードモードがあります。主に大容量ファイルを分割アップロードする必要がある場合や、ネットワークが不安定な環境でのアップロードに使用されます。OSS を例に、Multipart Upload のアップロードフローを以下に示します。

1. InitiateMultipartUpload：Multipart Upload モードでデータを転送する前に、このインターフェイスを呼び出して OSS に Multipart Upload イベントの開始を通知する必要があります。パラメーターとしてターゲットファイルのアドレスを指定し、uploadId を取得して後続のアップロードに使用します。

2. UploadPart：MultipartUpload を初期化した後、指定されたオブジェクト名と Upload ID に基づいてデータを分割アップロードできます。uploadPart インターフェイスを繰り返し呼び出して異なるデータパートをアップロードでき、並行呼び出しも可能です。

3. CompleteMultipartUpload：すべてのデータパートのアップロード完了後、CompleteMultipartUpload インターフェイスを呼び出してファイル全体の MultipartUpload を完了する必要があります。completeMultipartUpload の完了後、ファイルは OSS 上で外部から可視化されます。completeMultipartUpload が戻るまで、ファイルは外部から不可視です。

4. AbortMultipartUpload：AbortMultipartUpload インターフェイスは Multipart Upload イベントの終了に使用されます。CompleteMultipartUpload の前であれば、いつでも MultipartUpload を終了できます。

5. ListMultipartUploads：ListMultipartUploads は、現在実行中のすべての Multipart Upload イベント、つまり初期化済みでまだ Complete または Abort されていない Multipart Upload イベントを一覧表示するために使用されます。

Multipart Upload ベースの No-Rename Committer の実装
Multipart Upload 機能のサポートと S3/OSS ファイルシステムレベルのカスタマイズを組み合わせることで、データ整合性を確保した前提の下、Rename 操作を不要とする Job Committer を実装できます。具体的なジョブコミットプロセスは以下のとおりです。

1. setupJob：ジョブ一時ディレクトリを作成します。

2. setupTask：タスクの一時ディレクトリを設定します。タスク実行中、MultiUpload インターフェイスを使用してファイルをジョブの最終ディレクトリに直接書き込みます。ファイルクローズ時に CompleteMultipartUpload インターフェイスは呼び出さず、すべてのアップロードブロック情報をタスク一時ディレクトリ内のファイルに記録します。

3. commitTask：タスク一時ディレクトリ内の複数ファイルのアップロードブロック情報を 1 つのファイルに統合し、ジョブ一時ディレクトリに書き込みます。

4. abortTask：タスクの一時ディレクトリをクリーンアップし、AbortMultipartUpload インターフェイスを使用してタスクが書き込んだすべてのファイルを中止 (abort) します。

5. commitJob：ジョブ一時ディレクトリ内のすべてのアップロードブロック情報を読み取り、CompleteMultipartUpload インターフェイスを呼び出して、すべてのファイルの MultipartUpload を完了します。

6. abortJob：ListMultipartUploads を呼び出してタスクが書き込んだすべてのファイルを中止し、ジョブの一時ディレクトリをクリーンアップします。

タスクの実行中、Multipart Upload の関連インターフェイスを通じてアップロードとブロックデータの転送が初期化されますが、CompleteMultipartUpload が呼び出されるのは commitJob の時点です。Multipart Upload の特性により、CompleteMultipartUpload の呼び出し前はファイルが不可視であるため、データ整合性が保たれます。FileOutputCommitter と同様に、commitJob 時に複数のファイルの CompleteMultipartUpload を実行する必要があるため、データの不整合が発生する可能性のあるタイムウィンドウが存在します。ただし、ファイルのアップロードプロセス自体はタスク内で分散方式で既に完了しており、CompleteMultipartUpload はジョブドライバーでの commitJob 時に非常に軽量なリクエストとなるため、このタイムウィンドウは非常に短く、失敗の可能性も低く、大半のビジネスシナリオの要件を満たせます。FileOutputCommitter V1 と比較すると、ジョブコミット時の CompleteMultipartUpload のコストは Rename よりもはるかに低く、データの不整合が発生する可能性のあるタイムウィンドウもはるかに小さくなります。FileOutputCommitter V2 と比較すると、V2 はデータ整合性を保証しませんが、JindoOssCommitter はデータ整合性が必要なより多くのシナリオに適用できます。

パフォーマンス面では、この方式はタスク内で分散方式で並行してデータを OSS に書き込み、Rename 操作が不要です。FileOutputCommitter V1/V2 が必要な 2 回および 1 回の Rename 操作と比べても、大幅なパフォーマンス向上が実現されています。

まとめ

オブジェクトストレージシステムが一般的に提供する Multipart Upload 機能を活用して実現された No-Rename Committer は、FileOutputCommitter V1/V2 バージョンと比較して、データ整合性とパフォーマンスの両面で大幅に改善されています。MapReduce および Spark で S3/OSS にデータを書き込むシナリオでの使用が推奨されます。S3ACommitter は Hadoop コミュニティバージョン 3.1.2 で既に利用可能であり、JindoOssCommitter も Alibaba Cloud EMR 環境バージョン 2.5.0 以上でデフォルトで有効化されています。