2021年1月更新

S3が完全に整合性を持つようになったため、ディレクトリ一覧の不整合に関連する問題は解消されました。ただし、名前変更の問題は依然として存在します。ディレクトリの名前を変更して作業をコミットすることは、安全ではないだけでなく、非常に遅くなります。

このアーキテクチャドキュメントとコミッターは、S3が不整合だった時代に記述されました。2つのコミッターは、この問題に異なるアプローチで対処しました。

• ステージングコミッター：コミットするファイルのリストをワーカーからジョブマネージャ/ドライバに安全に伝播するために、クラスタHDFSファイルシステムに依存します。
• マジックコミッター：一貫性のあるオブジェクトストア上のディレクトリ一覧を提供するためにS3Guardを必要とします。

S3が整合性を持つようになったため、マジックコミッターは任意のS3バケットで安全に使用できます。したがって、どちらを使用するかは実験によって決定されます。

このアーキテクチャドキュメントは2017年に記述されたもので、S3Guardなどの追加の一貫性レイヤーを使用する場合にのみS3が整合性を持っていた時代のものでした。ドキュメントでは、当時存在していた要件/制約のうち、現在廃止されているものを示しています。

ジョブ設定

fs.mkdir(jobAttemptPath)

タスク設定

なし：ディレクトリは必要に応じて作成されます。

タスクコミット

タスク試行パスをタスクコミットパスに名前変更します。

def needsTaskCommit(fs, jobAttemptPath, taskAttemptPath, dest):
  return fs.exists(taskAttemptPath)


def commitTask(fs, jobAttemptPath, taskAttemptPath, dest):
  if fs.exists(taskAttemptPath) :
    fs.delete(taskCommittedPath, recursive=True)
    fs.rename(taskAttemptPath, taskCommittedPath)

実際のファイルシステムでは、これは `O(1)` のディレクトリの名前変更です。

模倣された名前変更を行うオブジェクトストアでは、コピーに `O(data)` 、すべてのファイルのリストと削除のオーバーヘッド（S3の場合、`（1 + files/500）` のリストと、同じ数の削除呼び出し）がかかります。

タスク中止

タスク試行パスを削除します。

def abortTask(fs, jobAttemptPath, taskAttemptPath, dest):
  fs.delete(taskAttemptPath, recursive=True)

実際のファイルシステムでは、これは `O(1)` の操作です。オブジェクトストアでは、ファイルのリストと削除にかかる時間に比例し、通常はバッチ処理されます。

ジョブコミット

すべてのタスクコミットパス内のすべてのファイル/ディレクトリを最終的な宛先パスにマージします。オプションで、宛先パスに0バイトの `_SUCCESS` ファイルを作成します。

def commitJob(fs, jobAttemptDir, dest):
  for committedTask in fs.listFiles(jobAttemptDir):
    mergePathsV1(fs, committedTask, dest)
  fs.touch("$dest/_SUCCESS")

(`mergePaths()` の詳細については以下を参照してください)

ジョブコミット中の障害は、クエリ全体を再実行する以外に回復できません。

def isCommitJobRepeatable() :
  return False

したがって、これはプロトコルにおける障害ポイントです。ファイル数が少なく、名前変更/リストアルゴリズムが高速な場合、脆弱性の期間は短くなります。大規模になると、脆弱性が増加します。コミットされたタスクの名前変更を並列実行することで、実際には削減できます。

ジョブ中止

ジョブ試行パス下のすべてのデータを削除します。

def abortJob(fs, jobAttemptDir, dest):
  fs.delete(jobAttemptDir, recursive = True)

ジョブクリーンアップ

def cleanupJob(fs, dest):
  fs.delete('$dest/_temporary', recursive = True)

`commitJob()` 前のジョブ復旧

すべてのコミッターについて、回復プロセスはアプリケーションマスターで行われます。1. 前回の試行のジョブ履歴ファイルがロードされ、成功したと記録されたタスクが判別されます。2. 成功した各タスクについて、ジョブコミッターは、前回の試行の詳細から構築された `TaskAttemptContext` を使用して、その `recoverTask()` メソッドが呼び出されます。3. メソッドが例外を発生させない場合、回復したと見なされ、再実行されません。4. その他のすべてのタスクは実行のためにキューに入れられます。

v1コミッターの場合、タスクの回復は簡単です。前回の試行からのコミットされたタスクのディレクトリは、現在のアプリケーション試行のディレクトリの下に移動されます。

def recoverTask(tac):
  oldAttemptId = appAttemptId - 1
  fs.rename('$dest/_temporary/oldAttemptId/${tac.taskId}',
    '$dest/_temporary/appAttemptId/${tac.taskId}')

これにより、ジョブドライバ（ここではMR AM）の障害からの回復時間が大幅に向上します。失われた作業は、データが生成されたがまだコミットされていない進行中のタスクの作業のみです。

ジョブドライバの障害のみがジョブの再起動を必要とし、個々のタスクは必要ありません。したがって、これは並列実行されるタスクの数とは無関係に、クエリの期間によって発生する確率です。

タスクが長いほど、障害のリスクが高くなり、進行中の作業を回復する価値が高まります。

高速クエリは障害のリスクが低いだけでなく、ジョブ全体を再実行するだけで障害から回復できます。これは、進行中のジョブを回復しようとしないSparkにおける暗黙的な戦略です。クエリが高速であるほど、回復戦略はシンプルになります。

`commitJob()` 中のジョブ復旧

これは不可能です。ジョブコミット中の障害により、宛先ディレクトリをクリーンアップした後、ジョブ全体を再実行する必要があります。

`mergePaths(FileSystem fs, FileStatus src, Path dest)` アルゴリズム

`mergePaths()` はデータをマージするためのコアアルゴリズムです。実装では、`mergePaths()` と `renameOrMerge()` の2つの相互再帰ルーチンにわたって両方のアルゴリズムの戦略が混在しているため、やや分かりにくくなっています。

ここでは、2つのアルゴリズムが分割され、相互再帰メソッドの1つがインライン化されています。

def mergePathsV1(fs, src, dest) :
  if fs.exists(dest) :
    toStat = fs.getFileStatus(dest)
  else:
    toStat = None

  if src.isFile :
    if not toStat is None :
      fs.delete(dest, recursive = True)
    fs.rename(src.getPath, dest)
  else :
    # src is directory, choose action on dest type
    if not toStat is None :
      if not toStat.isDirectory :
        # Destination exists and is not a directory
        fs.delete(dest)
        fs.rename(src.getPath(), dest)
      else :
        # Destination exists and is a directory
        # merge all children under destination directory
        for child in fs.listStatus(src.getPath) :
          mergePathsV1(fs, child, dest + child.getName)
    else :
      # destination does not exist
      fs.rename(src.getPath(), dest)

v2タスクコミット

タスク試行パスをタスクコミットパスに名前変更します。

def needsTaskCommit(fs, jobAttemptPath, taskAttemptPath, dest):
  return fs.exists(taskAttemptPath)

def commitTask(fs, jobAttemptPath, taskAttemptPath, dest):
  if fs.exists(taskAttemptPath) :
    mergePathsV2(fs, taskAttemptPath, dest)

v2タスク中止

タスク試行パスを削除します。

def abortTask(fs, jobAttemptPath, taskAttemptPath, dest):
  fs.delete(taskAttemptPath, recursive=True)

コスト：通常のファイルシステムでは `O(1)` 、オブジェクトストアでは `O(files)`。

v2ジョブコミット

すべてのタスク出力が既に完了しているため、必要なのは `_SUCCESS` マーカーにアクセスすることだけです。

def commitJob(fs, jobAttemptDir, dest):
  fs.touch("$dest/_SUCCESS")

コスト：`O(1)`

ジョブ中止中の障害は暗黙的に繰り返すことができます。

def isCommitJobRepeatable() :
  return True

v2ジョブ中止

ジョブ試行パス下のすべてのデータを削除します。

def abortJob(fs, jobAttemptDir, dest):
  fs.delete(jobAttemptDir, recursive=True)

コスト：通常のファイルシステムでは `O(1)` 、オブジェクトストアでは `O(files)`。

v2タスク復旧

データは宛先ディレクトリに書き込まれないため、タスク試行ディレクトリを削除することでタスクをクリーンアップできます。

`commitJob()` 前のv2ジョブ復旧

データは宛先ディレクトリに名前変更されているため、コミット済みとして記録されたすべてのタスクには、まったく回復は必要ありません。

def recoverTask(tac):

アクティブなタスクとキューに入れられたタスクはすべて実行のためにスケジュールされます。

`commitTask()` 中の障害と同じ弱点があります。そのコミット操作中に失敗したタスクを繰り返すのは、生成されたすべてのファイルの名前がすべてのタスク試行で一定の場合にのみ安全です。

タスク試行がコミットするように指示され、そのコミットが完了したと記録されていない間にジョブAMが失敗した場合、進行中のタスクの状態は不明です…実際には、この時点でジョブを回復することは安全ではありません。

`commitJob()` 中のv2ジョブ復旧

これは簡単です。`commitJob()` が再呼び出されます。

タスク設定

Task.initialize()：構成を読み込み、現在のジョブとタスクにバインドされたJobContextImplとTaskAttemptContextImplインスタンスをインスタンス化します。

タスクコミット

作業が完了すると、Task.done()が呼び出されます。これは基本的に次のコードパスです。

if (committer.needsTaskCommit(taskContext)) {
  // get permission to commit from AM
  int retries = MAX_RETRIES;
  while(true) {
    try {
      umbilical.commitPending(taskId, taskStatus);
      break;
    } catch(IOException ie) {
      if (--retries == 0) {
        // FAIL WITHOUT CALLING ABORT
        System.exit(67);
      }
    }
  }
  // commit the work
  try {
    committer.commitTask(taskContext);
  } catch (IOException iee) {
    // failure: abort
    try {
      committer.abortTask(taskContext);
    } catch (IOException ioe)  {
      LOG.warn("Failure cleaning up: " +
               StringUtils.stringifyException(ioe));
    }
    throw iee;
  }
}

つまり、書き込むデータがある場合に限り、タスクはAMからそうするためのクリアランスを要求します。これにより、別のタスクが書き込んだ場合に投機的作業がコミットされないこと、そして、タスクがAMとの連絡を失った場合でも、投機的ではない作業がコミットされないことが保証されます。

つまり、ジョブのAMが失敗した場合、またはタスクから別のネットワークパーティションにある場合、タスクの出力はコミットされません。

許可が与えられると、コミット呼び出しが呼び出されます。これが機能しない場合、abortTask()が呼び出されます。そこで発生したすべてのエラーはログに記録され、無視されます。

このメソッドは実際には、committer.abortTask()が常に呼び出されることを保証しないという点で制限されているようです。umbilical.commitPending()呼び出しが繰り返し失敗し、タスクプロセスが中止された場合、committer.abortTask()は呼び出されません。これがネットワークパーティションによるものである場合、AMがcommitter.abortTask()を呼び出し、AMの障害である場合は、再起動されたAMが以前の試行をクリーンアップできると期待できます。従来のFileOutputCommitterの場合、以前の試行のデータのリスト表示と削除は簡単です。ただし、マルチパートアップロードをコミットされていないデータをアップロードする手段として使用しているS3コミッタの場合、保留中のアップロードが常に中止されるようにすることが重要です。これは、次のように行うことができます。

• Task.done()内のすべてのタスク側エラーブランチがcommitter.abortTask()を呼び出すようにします。
• ジョブのコミットと中止によって、同じディレクトリツリーへの保留中のマルチパート書き込みがすべてクリーンアップされるようにします。つまり、他のジョブが同じツリーに書き込んでいないことを要求し、保留中のすべての操作をリストしてキャンセルします。
• パス下にコミットがないことを検証するCLIコマンドを追加します。コミットが存在する場合は、ゼロ以外の終了コードを返します。これはテストで使用できます。
• パス下のコミットをパージするCLIコマンドを追加します。これは、必要に応じてワークフローで使用でき、異なるテストの出力を分離するためにテスト実行で使用できます。
• システムの推奨されるパージ保留コミットタイムアウトを文書化します。
• 統合テストにプローブを追加して、書き込み後に保留中のアップロードが存在しないことを検証します。

Application MasterでのMapReduceによるコミッタの使用方法

AMはコミッタを使用してジョブを設定およびコミットします。AMの障害と回復をサポートするために、OutputCommitter.isRecoverySupported()を使用して、成功したタスクのすべての出力が最終ジョブで使用できるか、またはジョブ全体をリセットして繰り返す必要があるかを宣言します。OutputCommitter.isCommitJobRepeatable()は、もう1つの質問に対処します。コミッタはコミットプロセス自体の失敗から回復できますか。

コミッタはApplication Masterで作成され、org.apache.hadoop.mapreduce.v2.app.commit.CommitterEventHandlerのインスタンスに渡され、その後、ジョブレベルのライフサイクルを管理します。

MRv1 APIが使用されている場合、コミッタは"mapred.output.committer.class"の値から選択されます。MRv2 APIでは、出力形式がインスタンス化され、タスクとタスク試行ID 0を使用してコミッタが要求されます。コミッタはCommitter.getOutputCommitter(taskContext)への呼び出しを介して出力形式から取得されます。(ジョブ、0、0)タスクと試行IDを持つタスク試行コンテキストが再度使用されます。つまり、ジョブコミッタの場合でも、コミッタを要求する際に、常にタスクコンテキストが出力形式に渡されます。OutputCommitter.abortTask()のすべての実装について、コンテナとタスクを実行しているホストではなく、AMから実行できることが重要です。さらに、中止に必要なすべての情報（パス、ファイルシステムインスタンスなど）は、コンストラクタに渡されたコンテキストから初期化されたフィールドに依存するのではなく、メソッドに渡されたTaskAttemptContextから取得する必要があります。

private void touchz(Path p, boolean overwrite) throws IOException {
  fs.create(p, overwrite).close();
}

boolean commitJobIsRepeatable = committer.isCommitJobRepeatable(
      event.getJobContext());
try {
  touchz(startCommitFile, commitJobIsRepeatable);
  waitForValidCommitWindow();
  committer.commitJob(event.getJobContext());
  touchz(endCommitSuccessFile, commitJobIsRepeatable);
} catch (Exception e) {
  touchz(endCommitFailureFile, commitJobIsRepeatable);
}

AM：ジョブコミットのキャンセル

コミットを実行するスレッドが中断されます。CommitterEventHandlerはそれが終了するのを待ちます。（ミリ秒単位でyarn.app.mapreduce.am.job.committer.cancel-timeoutに設定されています）。

AM：タスクの中断

AMは、コンテナの失敗/損失を処理するときに、タスク試行コンテキストを使用してOutputCommitter.taskAbort()を呼び出す場合があります。つまり、コンテナの失敗時に、AMのコミッタを使用してAMでタスクの中断操作が実行されます。これにより、新しいコンテナを作成する必要がなくなり、「ベストエフォート」タスクの中断はコンテナの失敗にも対応できるようになります。

AMとタスクコンテナ間のパーティションは、パーティション化されたコンテナ内のタスクがまだ実行されている間に、このAMによって実行されるタスクの中断が発生する可能性があることを意味します。中断操作の後に出力の書き込み操作が失敗しない限り、パーティション化されたタスクは、talkToAMTGetPermissionToCommit()のTask.done()での独自のタスクコミットシーケンスまで、パーティションに気付かない可能性があります。

S3コミッタの要件

S3Aコミッタの設計要件は次のとおりです。

1. 最終的に整合性のあるS3オブジェクトストアを、S3Aファイルシステムクライアントを介した作業の信頼性の高い直接的な宛先としてサポートします。
2. 効率的：名前変更を行わず、ジョブドライバのタスクとジョブコミットフェーズの遅延を最小限に抑えることを意味します。
3. タスクの失敗と投機をサポートします。
4. 既存のコード（Hadoop MapReduce、Spark、Hive）で使用できます。
5. 既存のFileOutputFormatのサブクラスへのレトロフィット、または特定のFileOutputFormatを期待するコミッタとの互換性があります。
6. すべてのタスク試行、ジョブの以前のすべての試行、および以前の未完了のジョブからコミットされていないデータをクリーンアップします。
7. セキュリティ：同一ファイルシステムへの書き込みアクセス権を持つ他のユーザーからの権限昇格を許可しない。

コミットロジック

コアアルゴリズムは次のとおりです。

1. 出力の宛先ディレクトリ（例：FileOutputFormatとそのサブクラス）は、ローカルfile://参照です。
2. タスクコミットは、宛先オブジェクトストレージへのマルチパートPUTを開始します。
3. タスクの保留中のPUTのリストは、一貫性のあるクラスター全体のファイルシステム内の単一ファイルに永続化されます。Netflixでは、それがHDFSです。
4. 標準のFileOutputCommitter（アルゴリズム1）を使用して、これらのファイルのコミット/アボートを管理します。つまり、成功したタスクからのコミットするファイルのリストのみを（一時的な）ジョブコミットディレクトリにコピーします。
5. S3ジョブコミッターは、HDFSでコミットされた各タスクの保留中のファイルリストを読み取り、それらのPUTリクエストを完了します。

FileOutputCommmitterを使用してコミットするファイルのリストの伝播を管理することにより、既存のコミットアルゴリズムは、ジョブの最後にコミットされるファイルを暗黙的に定義するアルゴリズムになります。

Netflixの貢献には、S3のHadoopOutputCommitter実装があります。

主なクラスは3つあります。* S3MultipartOutputCommitterは、コミットロジックを処理する基本コミッタークラスです。これは直接使用しないでください。* S3DirectoryOutputCommitterは、競合解決を使用してS3に非パーティションデータを書き込みます。* S3PartitionedOutputCommitterは、競合解決を使用してS3にパーティションデータを書き込みます。

呼び出し元は、単一ディレクトリ出力にはS3DirectoryOutputCommitterを、パーティションデータにはS3PartitionedOutputCommitterを使用する必要があります。

これらのS3コミッターは、ローカルFSの一時ディレクトリにタスク出力を書き込むことによって機能します。タスク出力は、getTaskAttemptPathとgetWorkPathによってローカルFSに向けられます。

競合解決

単一ディレクトリとパーティションコミッターは、データをアップロードする前に、S3にターゲットパスが存在するかどうかを確認することで競合解決を処理します。fs.s3a.committer.staging.conflict-modeを設定することで制御される3つの競合解決モードがあります。

• fail：出力ディレクトリまたはパーティションが既に存在する場合は、タスクを失敗させます。（デフォルト）
• append：ディレクトリまたはパーティションが既に存在するかどうかを確認せずに、データファイルをアップロードします。
• replace：出力ディレクトリが存在する場合は、削除して新しいデータで現在のコンテンツを置き換えます。

パーティションコミッターは、ジョブの実行前ではなく、出力データとの競合が検出された場合に競合モードを適用します。競合解決は出力モードとは異なるため、競合がない場合はモードを適用しません。たとえば、パーティションを上書きする場合は、新しい出力が作成されるかどうかに関係なく、含まれるすべてのサブパーティションとデータを削除する必要があります。競合解決は、出力データを持つパーティションのみを置き換えます。

競合モードがreplaceの場合、競合するディレクトリはジョブコミット中に削除されます。データは、すべてのタスクが正常に完了した場合にのみ削除されます。

書き込みを識別するUUIDが、S3にアップロードされるファイル名に追加されます。これにより、ジョブコミット中に失敗したジョブからのデータのロールバックが可能になり（下記の失敗例を参照）、データを追加で既存のディレクトリに追加する場合のファイルレベルの競合が回避されます。

注記 存在のチェックは、宛先パスのS3AFileSystem.getFileStatus()リクエストを介して行われます。S3ストアのビューが一貫性がない限り、新しく削除されたオブジェクトがプローブでまだ検出される可能性があり、実際の競合がなくてもコミットが失敗する可能性があります。

パフォーマンス

これまでの提案（以降「マジック」コミッター）と比較して、このコミッター（「ステージング」コミッター）は、各タスクの最後に各ファイルをアップロードするという追加のオーバーヘッドを追加します。これはO(data)操作です。並列化できますが、コンピュートノードからS3への帯域幅と、S3の宛先シャードの書き込み/IOP容量によって制限されます。多くのタスクがほぼ同時に完了する場合、帯域幅のピーク負荷によりアップロードが遅くなる可能性があります。

コミット時間は同じになり、Netflixコミッターは既にここで並列化ロジックを実装しているため、O(files/threads)の時間になります。

回復力

タスクとジョブのコミットには、既に多くの障害処理コードが存在します。

コミットのいずれかの障害は、タスク/ジョブのコミットアクションのベストエフォートによる中止/ロールバックをトリガーします。

タスクコミットは、FileOutputCommitter に委譲して、1 つのタスクの出力だけがジョブコミットに到達するようにします。

同様に、タスクが中止された場合、ローカルFSの一時出力は削除されます。

コミッターの実行中にタスクが停止した場合、データがローカルFSに残ったり、S3に未完成の部分として残ったりする可能性があります。S3の未完成のアップロード部分はテーブルリーダーには表示されず、ターゲットバケットのライフサイクルポリシーのルールに従ってクリーンアップされます。

ジョブコミット中の障害は、既に完了しているファイルの削除と残りのアップロードの中止によって処理されます。アップロードは個別に完了するため、削除されたファイルはリーダーに表示されていました。

ジョブコミッターの実行中にプロセスが停止した場合、2 つの可能性のある障害が発生します。

1. 置き換えられるディレクトリの一部が削除されましたが、新しいデータは表示されません。
2. 新しいデータの一部が表示されますが、不完全であり、置き換えられたすべてのディレクトリは削除されています。完全なファイルのみが表示されます。

ジョブコミット中にプロセスが停止した場合、クリーンアップは手動プロセスです。ファイル名には書き込みごとにUUIDが含まれているため、ファイルの識別と削除が可能です。

タスク実行中の障害

すべてのデータはローカルの一時ファイルに書き込まれます。これらをクリーンアップする必要があります。

ジョブは、ローカル（保留中）データをパージする必要があります。

タスクコミット中の障害

アップロードプロセス中のプロセス障害により、保留中のマルチパートPUTのリストがクラスタファイルシステムに永続化されません。この期間はタスク実行全体よりも短くなりますが、特に大規模なアップロードでは、依然として無視できない可能性があります。

ファイルごとの永続化、またはアップロードリストの増分上書きによって、ここでの問題は軽減される可能性がありますが、それでも保留中のマルチパートアップロードが記録されないという小さなリスクが残ります。

タスクコミット前の明示的なタスク中止

タスクはすべてのローカルデータを削除します。アップロードは開始されません。

データアップロード中のS3との通信失敗

アップロードに失敗した場合、タスクは

• S3AFileSystemクラスとAWSライブラリに実装されているリトライポリシーを使用して再試行します。
• 最終的に：既に開始されている保留中のマルチパートアップロードの中止を試みます。
• ローカルFSの一時ファイルを削除します。

明示的なジョブ中止

実行中のすべてのタスクが中止され、クリーンアップされます。完了したすべてのタスクの保留中のコミットデータを読み込み、保留中のマルチパートPUTリクエストを中止できます。

これは、以下によって行われます。

1. ベストエフォートでの読み取りを試みた後、正常に読み取られたすべてのファイルの中止を伴う、すべてのローカルファイルのリスト作成。
2. 保留中のすべてのマルチパートアップロードのリスト作成と中止。

アクション＃2のため、アクション＃1は冗長です。これは、アクション＃2を構成可能なオプションにするオプションを残しておくため、（同時に実行される>1つのパーティション化されたコミットの使用例を処理するために必要になります）に残されています。

ジョブコミット前のジョブドライバ障害

ローカルデータはv1コミットアルゴリズムで管理されているため、ジョブの2回目の試行では、最初の試行の保留中のコミットデータすべてが復旧されます。これらのタスクは再実行されません。

これにより、ジョブの中止時に、個々のタスクの.pendingsetファイルを読み取り、それらのアップロードの中止を開始するために使用できることも保証されます。つまり、復旧されたジョブは、前のジョブの保留中の書き込みをクリーンアップできます。

クエリエンジンが複数のジョブ試行をサポートしていない場合、保留中のコミットデータは復旧されません。明示的な中止操作を開始する必要があります（これにはCLIコマンドを追加します）、または保留中のリクエストを自動的に削除するようにS3バケットを構成する必要があります。

ジョブコミット中のジョブドライバ障害

失敗したジョブコミットによって既に実行されたアップロードは永続化されます。実行されていないアップロードは保留されたままになります。

コミッターは現在、回復不可能であると宣言していますが、実際にはそうではない可能性があります。回復プロセスは次のいずれかになる可能性があります。

1. .pendingsetファイルからのすべてのジョブコミットを列挙します（:= コミット）。
2. 宛先パス下の保留中のすべてのアップロードをリストします（:= 保留中）。
3. 書き込まれたすべてのファイルをリストします（より良い整合性のために、既知のファイル名に対するGET呼び出しを使用）。
4. まだ完了していないすべてのファイルを識別します（コミット - 書き込み済み）。
5. 保留中のアップロードのセットが一致することを確認します（保留中 = （コミット - 書き込み済み））。

ここでの主な問題は、ジョブコミット時間競合解決プロセスの進捗状況です。プリコミットフェーズが完了したことを知っている場合にのみ、宛先ディレクトリのすべてのファイルが保持されることを確信できるため、それが完了したかどうかを判断する方法です。これは、「プリコミットが完了するまでアップロードはコミットされない」というルールに基づいて暗黙的に決定できます。1つ以上のアップロードが完了したことが判明した場合、プリコミットが完了したと推測できるため、ジョブを繰り返すことができます。

これは危険な領域です。現時点では、コミッターは回復不可能であると宣言しています。

タスクコミット前のアプリケーション全体の障害

データは、一時ディレクトリのローカルシステムに残ります。これはクリーンアップされない可能性があります。

1つ以上のタスクコミット後、ジョブコミット前のアプリケーション全体の障害

• 保留中の書き込みごとに、マルチパートPUTリクエストが保留されます。
• HDFSの一時ディレクトリには、既知の保留中のすべてのリクエストがリストされます。

>1つのタスク障害後のジョブ完了/中止

1. ジョブ完了ディレクトリにリストされている保留中のすべてのPUTデータを読み込み、キャンセルする必要があります。
2. 宛先ディレクトリへの他の保留中の書き込みを列挙して中止する必要があります。これは、出力が書き込まれる前にタスクが失敗した場合の状況をキャッチします。
3. ローカルディレクトリの保留中のすべてのデータを削除する必要があります。

問題：宛先ディレクトリはどうなりますか？上書きするかしないか？マージポリシーによって異なる可能性があります。

コアコンセプト：遅延PUTコミットのための新しい/変更された出力ストリーム

このアルゴリズムは、変更されたS3ABlockOutputStream出力ストリームを使用します。これは、最終的なclose()操作でアクティブなマルチパートアップロードをコミットする代わりに、独立したプロセスがアップロードを完了または中止できるように、S3リポジトリに十分な情報を保存します。

元々は、OutputStream.close()で、単一のPUTを実行するか、進行中のマルチパート書き込みを完了するかを選択していました。

マルチパートPUTが進行中の場合、ストリームは進行中のアップロードが完了するまで待ちます（送信された最終ブロックを含む）。その後、最終的なマルチパートコミット操作を作成してPUTします。パーツのリスト（とその順序）は、opt中に作成されています。

対照的に、遅延コミットファイルに書き込む場合

1. 小さな書き込みでも、マルチパート書き込みを必ず開始する必要があります。この書き込みは、ストリームの作成中に開始される可能性があります。

2. close()呼び出しで書き込みをコミットする代わりに、操作をコミットするために必要なすべての情報を含むS3リポジトリ内のパスにPUTを実行します。つまり、最終パス、マルチパートアップロードID、およびアップロードされたパーツの順序付きETagリストです。

ファイルが「特殊」であることを認識することは問題です。通常のcreate(Path, Boolean)呼び出しは、作成されているファイルが遅延コミットファイルであることを認識する必要があります。そのため、特別な新しいストリームを返します。

これは、「マジック」一時ディレクトリ名__magicを使用して行われ、このパス下に作成されたすべてのファイルは、ストリーム書き込みプロセス中に完了されないことを示します。パス下に作成されたディレクトリは引き続き作成されます。これにより、個々のジョブとタスクの試行ごとに、ジョブ固有およびタスク固有のディレクトリを作成できます。

たとえば、パターン__magic/${jobID}/${taskId}を使用して、その特定のタスクの最終ディレクトリへの保留中のコミットを保存できます。そのタスクがコミットされると、そのパスに保存されている保留中のすべてのコミットファイルが読み込まれ、最終的なアップロードのコミットに使用されます。

最終ディレクトリが/results/latestであるジョブを考えてみましょう。

ジョブjob_400_1のタスク01試行01の中間ディレクトリは次のようになります。

/results/latest/__magic/job_400_1/_task_01_01

これは一時ディレクトリとして返されます。

クライアントがファイル/results/latest/__magic/job_400_1/task_01_01/latest.orc.lzoの作成を試行すると、S3A FSは/results/latest/latest.orc.lzoの最終宛先でマルチパートリクエストを開始します。

データが出力ストリームに書き込まれると、個々のマルチパートPUT操作として増分的にアップロードされます。

close()時に、サマリーデータがファイル/results/latest/__magic/job400_1/task_01_01/latest.orc.lzo.pendingに書き込まれます。これには、アップロードIDとアップロードされたデータのすべての部分とETagが含まれます。

マーカーファイルも作成されるため、新しく作成されたファイルが存在することを確認するコードが失敗しません。1.これらのマーカーファイルの長さはゼロバイトです。1.HTTPヘッダーx-hadoop-s3a-magic-data-lengthで最終ファイルの全長を宣言します。1.getXAttr("header.x-hadoop-s3a-magic-data-length")を呼び出すと、アップロードされたデータのバイト数が含まれる文字列が返されます。

これは、Sparkの書き込み追跡コードが作成されたデータ量を報告できるようにするために必要です。

失敗ケース

パフォーマンス

アップロード時間は、今日のブロックアップロード（s3a.fast.upload=true）の出力ストリームのものです。書き込みを通して継続し、close()操作では、保留中のデータをアップロードし、保留中のアップロードが全て完了するのを待つ遅延があります。最終的な完了リクエストのオーバーヘッドはありません。まだデータがアップロードされていない場合、close()時間は、マルチパートリクエストの開始と最終的なputの時間になります。これは、ストリーム作成時に常にマルチパートIDを要求することで簡素化できるかもしれません。

各タスクをコミットする時間は`O(files)`です。タスク試行ディレクトリ内およびその下の全ての.pendingファイルがリストされ、その内容が読み取られ、集約された.pendingsetファイルがジョブ試行ディレクトリにPUTされます。その後、.pendingファイルは削除されます。

ジョブをコミットする時間は`O(files/threads)`になります。

ジョブ試行ディレクトリの全ての.pendingsetファイルを読み込む必要があり、ファイルにリストされている不完全なアップロードごとにPUTリクエストを発行する必要があります。

[廃止] 全ての子要素の一括リスト作成で完全な整合性が要求されることに注意してください。代わりに、コミットするファイルのリストを、Sparkコミットプロトコルが許可するように、タスクからジョブコミッタに返すことができれば、不整合なオブジェクトストアにデータをコミットできます。

コスト

不完全なマルチパートアップロードにおけるコミットされていないデータは、S3バケットのストレージコストで課金されます。コストを抑えるために、失敗したジョブからの未処理のデータを削除する必要があります。これは、S3バケットのライフサイクルポリシー、または作成する必要があるコマンドツールを通じて行うことができます。

このアルゴリズムの限界

1. ファイルは存在しなくなるため、close()呼び出し後には表示されません。保留中のコミットファイルを認識していたコードは失敗します。

2. タスクとジョブの失敗により、未処理のマルチパートアップロードが残ります。これらはガベージコレクションする必要があります。S3は現在自動クリーンアップをサポートしています。S3Aは起動時に実行するオプションがあり、hadoop s3コマンドで呼び出し元が明示的に実行できるようにする予定です。タスクが書き込みの開始時に書き込みのアップロードIDを明示的に書き込む場合、ジョブコミッタによるクリーンアップが可能になります。

3. 非常に小さなファイルを書き込む時間は、PUTとCOPYよりも長くなる可能性があります。これは本番環境では関連がないため、この問題は無視しています。小さなファイル操作を最適化しようとする試みは、開発、保守、テストを複雑にするだけです。

4. 一時情報を含むファイルは、実際のデータと間違われる可能性があります。

5. 問題の原因の診断が難しくなる可能性があります。特に、一時ディレクトリのディレクトリ構造のデバッグレベルのリスト作成によって、多くのログが役立ちます。

6. 保留中のPUTリクエストを確実にリストするには、リストの一貫性が必要です。S3エンドポイントの一貫性を確実に特定する方法がない場合、データの損失を伴う最終的に一貫性のあるストアを依然として使用する場合があります。

7. 同時に同じ宛先ディレクトリに書き込むジョブが複数ある場合、出力が混乱する可能性があります。これは、現在のコミットアルゴリズムでは今日でも当てはまります。

8. 同じS3Aクライアント/タスク内で、シーケンシャルまたは並列のいずれかで、同じ宛先ファイルに書き込むクライアントを複数作成できます。

9. [廃止] 一貫性のあるメタデータストアであっても、ジョブが既存のファイルを上書きする場合、更新がデータの全てのレプリカに伝播するまで、古いデータはデータを読み取るクライアントに表示される可能性があります。

10. 操作がディレクトリのコンテンツを完全に上書きしようとしている場合、機能しません。既存のデータはクリーンアップされません。クリーンアップ操作をジョブコミットに含め、上書きされていない宛先ディレクトリ内の全てのファイルを削除する必要があります。

11. 保留中のコミットデータの保存以外には使用できない__magicなどのパス要素が必要です。

12. 全てのFS操作にコードを追加しない限り、__magicツリー下のファイルを操作することは依然として可能です。それは悪いことではありませんが、混乱する可能性があります。

13. 書き込まれたデータはコミットされるまでマテリアライズされないため、どのプロセスも、作成したばかりのファイルを読み取ったり、操作したりすることはできません。

S3ABlockOutputStreamへの変更

S3ABlockOutputStreamをコピーアンドペーストする必要を避けるために、アップロードの即時/遅延完了を管理するPutTrackerクラスを使用して構築されるように変更されました。適切なポイントで呼び出されます。

• 初期化、マルチパートアップロードをすぐに開始するかどうかを示すマーカーを返します。
• マルチパートPUT初期化。
• 単一put初期化（このアルゴリズムでは使用されませんが、完全性のために役立ちます）。
• ブロックアップロード初期化、失敗、完了（関連するスレッドから）。
• close()に入力しました。全てのブロックが完了しました—保留中のマルチパートをコミットする必要があるかどうかを示すマーカーを返します。
• abort()呼び出しでのマルチパートアボート（おそらく：コアロジックを他の場所に移動する）。

基本実装は、close()呼び出しでMPUを実行する必要があることを宣言すること以外は何もしません。

S3Aコミッタバージョンは、1. 初期化中に常に開始する 1. close()操作で、後でコミットするために必要なデータを全て保存します。

org.apache.hadoop.mapred.FileOutputFormatによるMRv1サポート

FileOutputCommitterをサブクラス化しないことのデメリットは、MRv1 API呼び出しのプロトコルをFileOutputCommitterにラップして中継するために使用されるコードが機能しなくなることです。新しいコミッターは検出されません。

これはSparkで確認できます。Sparkでは、V1 APIはRDDクラス（RDD.saveAsHadoopFile()）からエクスポートされています。後継コードであるPairRDDFunctions.saveAsNewAPIHadoopFile()は機能します。*オブジェクトストレージで高性能なコミットを実現するには、MRv2コミットプロトコルを使用する必要があります。つまり、V2クラスを使用する必要があるということです。

すべてのコミッターのセキュリティリスク