問題タブ [input-split]

次のようなバイナリ形式で記述された大規模な時系列データ ファイルを処理する M/R ジョブを作成しています (読みやすくするためにここに新しい行があり、実際のデータは明らかに連続しています)。

ここで、timestamp は単に 8 バイトの構造体であり、最初の 2 バイトで識別できます。実際のデータは、上に表示されているように、重複した値のタイムスタンプの間に境界があり、1 つ以上の事前定義された構造体が含まれています。キーと値のペアをマッパーに発行するカスタム InputFormat を作成したいと思います。

論理的には、TIMESTAMP現在TIMESTAMPの<TIMESTAMP, DATA>. 私の問題は、 内の分割間の同期でRecordReaderあるため、特定のリーダーが次の分割を受信した場合

これにアプローチする良い方法は何ですか？CustomRecordReader分割間で同期でき、データを失わないように、ファイル オフセットにアクセスする簡単な方法はありますか? 分割の処理方法について概念的なギャップがあると思うので、これらの説明が役立つかもしれません。ありがとう。

map.input.startプロパティは、行の先頭の位置を教えてくれないようです（もちろん、最初のmap.input.startが0である場合を除きます）。map.input.startは、マッパーの入力の最初の行の途中にある場合もあれば、前のマッパーの入力の最後の行の途中にある場合もあります。これは予想されることですか？もしそうなら、どうすれば行のバイトオフセットを取得できますか？マッパーへのキーを破棄するHadoopストリーミングを使用しているため、使用TextInputFormatは機能しません

FileSplit対応するファイルから実際のバイトを読み取るためにオブジェクトを使用するクライアント アプリケーションを想定します。

そのためには、次のようなコードを使用InputStreamして からオブジェクトを作成する必要があります。FileSplit

-1 によるストリームの調整は、Hadoop MapReduce LineRecordReaderクラスなどのいくつかのシナリオに存在します。ただし、FSDataInputStream seek()メソッドのドキュメントには、特定の場所をシークした後、次の読み取りはその場所から行われることが明示的に示されています。これは、上記のコードが 1 バイトずれている (?) ことを意味します (?)。

問題は、すべての InputSplit 読み取りケースで「-1」の調整が必要かどうかです。

ところで、FileSplit正しく読みたい場合は、その開始点までシークするだけでは十分ではありません。これは、すべての分割にも、実際の HDFS ファイルの終了点とは異なる終了点がある可能性があるためです。したがって、対応するInputStreamものは「制限」されている必要があります。つまり、次のように最大長があります。

この場合、fsin上で「ネイティブ」ストリームを作成した後、org.apache.commons.io.input.BoundedInputStreamクラスを使用して「境界」を実装します。

アップデート

どうやら調整は、LineRecordReader最後の行全体を確実に読み取るために分割の境界を超えるクラスの 1 つのユース ケースにのみ必要です。

これに関するより詳細な議論は、以前の質問とMAPREDUCE-772のコメントにあります。

私は次のwikiを読みましたが、まだ1つのことを明確にすることができません.

https://wiki.apache.org/hadoop/HadoopMapReduce

たとえば、2 つの HDFS ブロックに分割された大きなファイルがあり、ブロックが 2 つの異なるマシンに物理的に保存されているとします。両方のブロックをローカルにホストするノードがクラスター内にないことを考慮してください。私が理解したように、TextInputFormat HDFS ブロック サイズは通常、分割サイズと同じです。2 つの分割があるため、ブロックをローカルに保持する 2 つの別々のマシンで 2 つのマップ インスタンスが生成されます。ここで、HDFS テキスト ファイルが行の途中で壊れてブロックを形成したとします。Hadoop はブロック 2 を 2 番目のマシンから最初のマシンにコピーして、2 番目のブロックから最初の行 (壊れた半分) を提供し、最初のブロックの最後の壊れた行を完成させることができますか?

以下のようなログファイルがあります

これを処理するために mapreduce を使用したいと思います。そして、インシデントID、ステータス、およびインシデントにかかった時間を抽出したい

可変レコード長を持つ両方のレコードを処理する方法と、レコードが終了する前に入力分割が発生した場合はどうなりますか。

「Hadoop - Definitive Guide」には、次のように書かれています -->

ジョブを実行しているクライアントは、getSplits() を呼び出してジョブの分割を計算し、それらを jobtracker に送信します。jobtracker は、それらのストレージの場所を使用して map タスクをスケジュールし、それらを tasktracker で処理します。

getLocations() がホスト名の配列を返すことはわかっています。

質問 1: クライアントはどのホスト名を返すかをどのように認識していますか? ジョブトラッカーの仕事じゃないの？

質問 2: 2 つの異なる InputSplit オブジェクトが同じホスト名を返すことはありますか? ホスト名はどのように決定されますか。誰がそれをしますか？

クライアントがファイルのすべてのホスト名 (レプリカを含む) を取得するために namenode に接続し、入力分割ごとに設定された場所に到達するためにいくつかの計算を行っているように感じます。本当ですか？

Wordcount では、投機的実行をオフにすると、ブロックごとに複数のマップ タスクを取得できるようです。

ジョブトラッカーは内部で何らかの魔法をかけて、InputSplits によって提供される以上のタスクを分散しますか?

をカスタマイズする必要がある入力ファイルがありますRecordReader。ただし、ここでの問題は、データがさまざまな入力分割に分散され、別のマッパーが最初のマッパーによって消費されるデータを取得する可能性があることです。

例:
ABCD
$ EF

先頭の「$」は、前の行の続きであることを意味します。

考えると、2 番目の分割は $ から始まります。さて、私の最初のマッパーは、最初の行の続きに何かがあることを知りません。また、データに 2 行目がまったくない可能性が非常に高いことにも注意してください。したがって、2行目を読むまで、または読まない限り、データの続きに何かがあるとは言えません。

この問題の解決策を見つけてください。

次のシナリオを検討してください。

それぞれ6 MBの4つのファイルがあります。HDFSブロックサイズは 64 MB です。

1blockは、これらすべてのファイルを保持します。余分なスペースがあります。新しいファイルが追加された場合、ここに収容されます

によってジョブのinput splitsが計算される場合 (通常は、各分割を処理のためにメモリにロードできるため、シーク時間が短縮されます。)Map-reduceInput formatsplit sizeHDFS block size

ここで行われる入力分割の数:

1. 4つのファイルすべてがaに含まれているため、1つblockですか？

2. それともファイルごとに1つの入力分割ですか?

3. これはどのように決定されますか？すべてのファイルを単一の入力分割として処理したい場合はどうすればよいですか?

NLineInputFormat の getSplitsForFile() fn を調べました。入力ファイル用に InputStream が作成され、その反復と分割が n 行ごとに作成されることがわかりました。効率的ですか？特に、この読み取り操作がマッパー タスクを起動する前に 1 つのノードで発生している場合。5 GB のファイルがある場合はどうなりますか。基本的には、ファイル データが 2 回シークされることを意味します。これがボトルネックである場合、hadoop ジョブはこれをどのようにオーバーライドしますか?

私のユースケースを clément-mathieu に提供するための編集

私のデータセットは、それぞれ約2GBの大きな入力ファイルです。ファイルの各行は、データベースのテーブル (私の場合は cassandra) に挿入する必要があるレコードを表します。データベースへの一括トランザクションを n 行ごとに制限したいと考えています。nlineinputformat を使用してこれを行うことに成功しました。私の唯一の懸念は、本番環境で現れる可能性のある隠れたパフォーマンスのボトルネックがあるかどうかです。