によるHadoop - The Definitive Guide
FileInputFormats が定義する論理レコードは、通常、HDFS ブロックにうまく収まりません。たとえば、TextInputFormat の論理レコードは行であり、HDFS の境界を頻繁に越えます。これは、プログラムの機能には関係ありません。たとえば、行が欠落したり壊れたりすることはありませんが、知っておく価値はあります。入力データ) は、いくつかのリモート読み取りを実行します。これにより発生するわずかなオーバーヘッドは、通常は重要ではありません。
レコード行が 2 つのブロック (b1 と b2) に分割されているとします。最初のブロック (b1) を処理するマッパーは、最後の行に EOL セパレータがないことに気付き、次のデータ ブロック (b2) から残りの行をフェッチします。
2 番目のブロック (b2) を処理するマッパーは、最初のレコードが不完全であり、ブロック (b2) の 2 番目のレコードから処理を開始する必要があることをどのように判断しますか?
興味深い質問です。詳細についてコードを調べるのに時間を費やしました。ここに私の考えがあります。分割は によってクライアントによって処理されるInputFormat.getSplitsため、FileInputFormat を見ると次の情報が得られます。
max(minSize, min(maxSize, blockSize))します。maxSizemapred.max.split.sizeminSizemapred.min.split.sizeFileSplit上で計算した分割サイズに基づいて、ファイルを異なる に分割します。ここで重要なのは、それぞれが入力ファイルのオフセットに対応するパラメーターFileSplitで初期化さstartれることです。その時点ではまだ回線の処理はありません。コードの関連部分は次のようになります。
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(new FileSplit(path, length-bytesRemaining, splitSize,
blkLocations[blkIndex].getHosts()));
bytesRemaining -= splitSize;
}
その後、LineRecordReaderによって定義されているを見ると、TextInputFormatそこで行が処理されます。
LineRecordReader化である a をインスタンス化しようとします。2 つのケースがあります。LineReaderFSDataInputStreamCompressionCodec定義されている場合、このコーデックは境界の処理を担当します。おそらくあなたの質問には関係ありません。ただし、コーデックがない場合は、興味深い点です。 のstartがInputSplit0 以外の場合、 1 文字バックトラックしてから、遭遇した最初の行をスキップし、\n または \r\n (Windows) で識別されます! 行の境界が分割の境界と同じ場合、有効な行をスキップしないようにするため、バックトラックは重要です。関連するコードは次のとおりです。
if (codec != null) {
in = new LineReader(codec.createInputStream(fileIn), job);
end = Long.MAX_VALUE;
} else {
if (start != 0) {
skipFirstLine = true;
--start;
fileIn.seek(start);
}
in = new LineReader(fileIn, job);
}
if (skipFirstLine) { // skip first line and re-establish "start".
start += in.readLine(new Text(), 0,
(int)Math.min((long)Integer.MAX_VALUE, end - start));
}
this.pos = start;
したがって、分割はクライアントで計算されるため、マッパーを順番に実行する必要はありません。すべてのマッパーは、最初の行を破棄する必要があるかどうかを既に認識しています。
したがって、基本的に、同じファイルに 100Mb ごとに 2 行ある場合、簡単にするために、分割サイズが 64Mb であるとします。次に、入力分割が計算されると、次のシナリオになります。
Map Reduceアルゴリズムは、ファイルの物理ブロックでは機能しません。論理入力分割で機能します。入力分割は、レコードが書き込まれた場所によって異なります。レコードは 2 つのマッパーにまたがる場合があります。
HDFSのセットアップ方法では、非常に大きなファイルを大きなブロック (たとえば、128 MB の測定) に分割し、これらのブロックの 3 つのコピーをクラスター内の異なるノードに格納します。
HDFS は、これらのファイルの内容を認識しません。レコードはBlock-aで開始されている可能性がありますが、そのレコードの終わりはBlock-bに存在する場合があります。
この問題を解決するために、Hadoop は、入力分割と呼ばれる、ファイル ブロックに格納されたデータの論理表現を使用します。MapReduce ジョブ クライアントは、入力分割を計算するときに、ブロック内の最初のレコード全体が開始する場所と、ブロック内の最後のレコードが終了する場所を特定します。
キーポイント:
ブロック内の最後のレコードが不完全な場合、入力分割には、次のブロックの位置情報と、レコードを完了するために必要なデータのバイト オフセットが含まれます。
下の図を見てください。
この記事と関連する SE の質問をご覧ください: Hadoop/HDFS ファイル分割について
Map-Reduce フレームワークは、ジョブの InputFormat に依存して次のことを行います。
InputSplit[] getSplits(JobConf job,int numSplits) は、これらを処理するための API です。InputFormat実装されたgetSplits() メソッドを拡張するFileInputFormat 。grepcodeでこのメソッドの内部を見てください。
私はそれを次のように見ています: InputFormat は、データの性質を考慮して、データを論理的な分割に分割する責任があります。
これを妨げるものは何もありませんが、ジョブにかなりの待ち時間が追加される可能性があります。必要な分割サイズの境界に関するすべてのロジックと読み取りは、ジョブトラッカーで行われます。
最も単純なレコード対応の入力形式は TextInputFormat です。次のように機能しています（コードから理解している限り）-入力形式は、行に関係なくサイズごとに分割を作成しますが、LineRecordReaderは常に：
a）そうでない場合、分割の最初の行（またはその一部）をスキップします最初の分割
b) 最後に分割の境界の後の 1 行を読み取ります (データが使用可能な場合、最後の分割ではありません)。
私が理解したことからFileSplit、最初のブロックに対して が初期化されると、デフォルトのコンストラクターが呼び出されます。したがって、start と length の値は最初はゼロです。最初のブロックの処理が終了するまでに、最後の行が不完全な場合、長さの値は分割の長さよりも大きくなり、次のブロックの最初の行も読み取られます。このため、最初のブロックの start の値は 0 より大きくなり、この状態では、LineRecordReaderは 2 番目のブロックの最初の行をスキップします。（ソースを参照）
最初のブロックの最後の行が完了すると、長さの値は最初のブロックの長さと等しくなり、2 番目のブロックの開始の値はゼロになります。その場合、LineRecordReaderは最初の行をスキップせず、最初から 2 番目のブロックを読み取ります。
理にかなっていますか？
コンストラクターの LineRecordReader.java の Hadoop ソース コードから: いくつかのコメントが見つかりました:
// If this is not the first split, we always throw away first record
// because we always (except the last split) read one extra line in
// next() method.
if (start != 0) {
start += in.readLine(new Text(), 0, maxBytesToConsume(start));
}
this.pos = start;
このことから、hadoop は分割ごとに余分な行を 1 つ読み取り (現在の分割の終わりに、次の分割で次の行を読み取る)、最初の分割でない場合は最初の行が破棄されると考えています。行レコードが失われて不完全にならないように
マッパーは通信する必要はありません。ファイル ブロックは HDFS にあり、現在のマッパー (RecordReader) は行の残りの部分を含むブロックを読み取ることができます。これは舞台裏で行われます。