問題タブ [recordreader]

RawLocalFileSystem の入力ストリームの getPos は、基になるストリームが閉じている場合に null ポインター例外をスローする可能性があることがわかりました。

カスタム レコード リーダーで遊んでいるときに、これを発見しました。

パッチを適用するには、「stream.available()」の呼び出しで例外がスローされるかどうかを確認し、例外がスローされる場合は getPos() 関数で 0 を返します。

既存の getPos() 実装は次の場所にあります。

https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20/src/examples/org/apache/hadoop/examples/MultiFileWordCount.java

RecordReader での getPos() の正しい動作は何ですか?

指定されたディレクトリでアーカイブを受信し、アーカイブを開いてその中のファイルを処理する mapreduce ジョブを定期的に起動するシステムがあります。次回同じアーカイブを再処理するのを避けるために、RecordReader の close() メソッドにフックして、最後のエントリが読み取られた後に削除されるようにします。

このアプローチの問題点は (私たちが考えている)、特定のマッピングが失敗した場合、別のマッピングを試みた次のマッパーが、元のファイルがレコード リーダーによって最初のファイルから削除されていることを発見し、爆発することです。すべてのマッピングと縮小が完了するまで待ってから、入力アーカイブを削除するのがよいと思います。

これはこれを行うための最良の方法ですか？

もしそうなら、システムが見つけたすべての入力ファイルのリストをメインプログラムから取得するにはどうすればよいでしょうか? （入力ディレクトリ全体をスクラブすることはできません。新しいファイルが存在する可能性があります）

すなわち:

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

例:
ABCD
$ EF

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

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

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

Hadoop v2 では、HDFS に格納されているいくつかの大きなバイナリ形式に基づいてRecordReaderおよび/またはを作成する必要があります。InputFormatファイルは基本的に、次の構造を持つ連結されたレコードです。

したがって、すべての境界点を知るには、ファイル全体をスキャンする必要があります。

このような構造に対処するカスタム リーダー/フォーマットの例はありますか?

すべての分割ポイントを事前に計算することを避けたいと思っています。データのループを無駄にする必要がないように、マッパーが必要とするときに各レコードをストリーミングしたいと考えています。ただし、分割ポイントを事前に計算する必要がある場合でも、カスタムスプリッターの作成方法がわからないため、可能であればそのようなものへのポインターもいただければ幸いです。

1 つの注意点: 各レコードの「ペイロード」は基本的に任意のバイナリ データであり、"FOOO"私の知る限り、4 バイトの定数が含まれている可能性があります。したがって、入力分割がレコードの途中で発生した場合、必ずしも"FOOO"次のレコードを見つけるためにの次のインスタンスに進むことはできません。ヘッダーだけでなくデータを検索し、必要な場所を探します。

私は最近Hadoopをいじり始め、pdfを処理するための独自の入力フォーマットを作成しました。

何らかの理由で、私のカスタム RecordReader クラスには初期化メソッドが呼び出されていません。(デバッグ環境が整っていないのでsysoutで確認)

Windows 7 32ビットでhadoop 2.2.0を実行しています。Hadoop jar は windows の下で盗聴されているため、yarn jar を使用して呼び出しを行っています...

ファイルのグループで構成される Map タスクの分割を作成するために、カスタムの結合ファイル入力形式を実装しました。スプリットの各ファイルをレコードリーダーに渡すソリューションを作成しましたが、すべて問題ありません。今、マップ関数にファイルのセット全体を渡そうとしています。

これは私のレコードリーダーコードです:

このコードでは、マップ関数がキーと値のベクトルを正しく受け取りますが、繰り返し発生します。つまり、 map 関数は 1 回呼び出されると思っていましたが、代わりに複数回呼び出されました。私は何を間違っていますか？

オブジェクト (クリック) がデータセットに表示される頻度をカウントする Hadoop ジョブを実装しようとしています。したがって、カスタムファイル入力形式を作成しました。レコード リーダーは、指定されたファイルの最初の行のみを読み取り、入力ストリームを閉じるようです。

コードは次のとおりです。

Pojo クラス:

FileInputFormat クラスは次のとおりです。

マッパークラス:

パーティショナー クラス:

サーブレット コンテナー内の Restful Web サービス呼び出しを介してトリガーされる Hadoop ジョブですが、これは問題にはなりません。

次のデータセット (例):

プログラムを実行すると、syso は次のように表示されます。

そのことから、レコードリーダーは最初の行のみを読み取ると結論付けています。なぜこの問題が発生し、どのように修正されますか?

私はHadoopの初心者です。一度に3行を読み取り、3行の入力がマッパーに与えられた回数を出力するこのカスタムRecordReaderプログラムに出くわしました。

RecordReader が使用される理由は理解できますが、入力形式クラスが基本的に mapreduce.TextInputFormat クラスを拡張している場合に、各 InputSplit に 3 行を含める方法を確認できません。そして、私の理解によると、TextInputFormat クラスは各行 (各 \n) に対して 1 つの InputSplit を発行します。

では、RecordReader はどのようにして各 InputSplit から 3 行を読み取ることができるのでしょうか? これがどのように可能かを誰かが説明してください。前もって感謝します！