問題タブ [mrjob]

3 つのステップで構成される mrjob があります。2 番目のステップでは、最初のステップの結果に加えて、S3 からのコンテンツが入力として期待されます。

最初のステップでいつでも「ストリーミング」できることを理解しています。つまり、エミットはそのままで、2番目のステップでのみ使用できますが、これは避けたいと思います。

mrjob の後のステップへの追加入力を定義する方法はありますか?

次のような記録があります。

それらは複数行ですが、必ずしも同じ行数ではなく、通常は改行で区切られています。以下の出力にどのように変換しますか？

Apache Pig は各行をレコードとして扱うため、このタスクには適していません。複数行レコードの処理に関するこのブログ投稿は知っていますが、より簡単な解決策がある場合は、Java について掘り下げることは避けたいと思います。Hadoop Streaming (またはmrjobのようなフレームワーク) を使用してこれを解決する方法はありますか?

処理するのに約 10 GB の比較的大きなファイルがあります。MRJob が RAM などでソートすることを決定した場合、ラップトップの RAM に収まらないのではないかと思います。

同時に、hadoop や EMR をセットアップしたくありません。仕事は緊急ではなく、就寝前に簡単にワーカーを起動して、翌朝結果を取得できます。言い換えれば、私はローカルモードに非常に満足しています. パフォーマンスが完璧ではないことはわかっていますが、今のところは問題ありません。

では、このような「大きな」ファイルを 1 台の弱いマシンで処理できるのでしょうか? はいの場合-何をすることをお勧めしますか（すぐに使い果たされるRAMディスクではなく、ファイルシステムを指すようにカスタムtmpディレクトリを設定する以外に）。バージョン 0.4.1 を使用すると仮定しましょう。

私はmrjobプログラムを持っていて、SQLデータベースからデータを取得するだけなので、ローカルファイルや入力ファイルを読み取る必要はありませんが、mrjobは「STDINから読み取る」ことを強制するので、入力ファイルとして空のファイルを作成するだけです. 入力ファイルなしでジョブを実行する方法はありますか?

Hadoop 用の単純な k-means クラスタリング コードを作成しました (2 つの別個のプログラム - マッパーとリデューサー)。コードは、ローカル ボックスの 2 次元点の小さなデータセットで動作しています。これは Python で書かれており、ストリーミング API を使用する予定です。

マッパーとリデューサーを実行するたびに、新しいセンターが生成されます。これらの中心は、次の反復の入力です。

提案に基づいて、マルチステップに適した mrjob,job python を使用しました。

これは 1 回の反復にすぎません。新しいセンターが生成された後にマッパーにフィードバックする方法を教えてください。つまり、最後のステップ (「リデューサー」) でわかるように、新しい中心が生成され、それをマッパー (最初のステップ) に再度フィードバックして、新しい中心との新しい距離を計算するということです。満足のいく収束。

(Mahout、spark、またはその他の実装については教えないでください。私はそれらを認識しています。)

MRJob スクリプトを実行し、CLI を使用して作業のために EMR クラスターを起動するとき、S3 からクラスター内の HDFS にデータをロードする方法を見つけようとしています。これをセットアップ プロセスの一部として実行したいと考えています。

S3 データを HDFS にプリロードするために MRJob で使用する適切な構文を理解するために、さまざまな場所を検索しました。私はまだ専門用語とプロセスについて理解を深めています。

MRJob は、ドキュメントで JarStep と Hadoop ストリーミングについて言及しています: https://pythonhosted.org/mrjob/guides/writing-mrjobs.html

Yelp には、これに関するディスカッション スレッドがあります: https://github.com/Yelp/mrjob/pull/665

AWS には、ロード用の構文を提供するツール S3DistCp があり、例は Java に関連しているように見えます。(2 つ以上のリンクを投稿することはできませんが、Google 検索でこれに関する詳細情報が得られます)

私のアプローチは、MRJob スクリプトのステップ メソッドに MRStep の前のステップとして JarStep を含めることでした。S3DistCp コマンドを引数として渡そうとしました。

引用符または入力と出力を分割する引用符、またはコンマ間のスペースを使用して、または使用せずに、これについていくつかのバリエーションを試しました。主に発生するエラーは、「スレッド 'main' java.lang.RuntimeException の例外: 引数 --src,, --desc,, が一致しません。

私が見つけた例は、次のようなものです。

JarStep.INPUT を置き換えるか、別の場所で定義する必要があるかどうかは明確ではありません。S3DistCp ドキュメントに合わせて --jobconf の下の CLI で Java コマンドを渡す必要があると考え始めています。

アイデアや単純な例があれば、私の方法で送ってください。

IPython ノートブックから mrjob の例を実行しようとしています

次に、コードで実行します

エラーが発生します：

IPython ノートブックから mrjob を実行する方法はありますか?

カスタム データで mrjob の Web サイトから基本的な例を実行しようとしています。ストリーミングを使用して Hadoop map reduce を正常に実行しました。また、Hadoop を使用せずにスクリプトを正常に試行しましたが、次のコマンドで mrjob を介して Hadoop で実行しようとしています。

mapred.py のソース コードは次のとおりです。

残念ながら、次のエラーが発生します。

mrjob は HDFS にディレクトリを作成できないようですが、残念ながらこの問題を解決する方法がわかりません。

私の Hadoop は cloudera CDH5.1 です。クイックスタート。

提案やアドバイスをお寄せいただきありがとうございます。

編集：

cloudera CDH4.7 で同じコードを実行しようとしました。クイックスタートでうまくいきました。したがって、私の修正された質問は次のとおりです。cloudera CDH5.1 です。mrjob フレームワークでサポートされていますか? もしそうなら、それを実行する方法は？

mapreduce について教えられたとき、重要なコンポーネントの 1 つはコンバイナーでした。マッパーが出力するデータの行数を減らすために、基本的にマップフェーズの最後にリデューサーを実行するのは、マッパーとリデューサーの間のステップです。処理する必要があるデータのサイズが (数テラバイト規模で) 増加するにつれて、reduce ステップは非常に遅くなります。私の友人と話したところ、これは彼の経験でもあり、コンバイナーを使用する代わりに、reduce ステップで各キーに送られる値の数を減らすハッシュ関数を使用して、reduce キーを分割すると言いました。 . 私はこれを試してみましたが、うまくいきました。コンバイナーステップがうまくスケーリングしないという経験をした人はいますか?なぜこの問題と回避策のドキュメントが見つからないのですか? 私'

[編集] コンバイナーよりもはるかに高速に動作する、私の友人が提案した回避策の例を次に示します。

出力する代わりにword, count

マッパー出力(word, hash(timestamp) % 1024), count

次に、マッパーの結果をマージするための 2 つの削減ステップがあります。

MRJob を使用して、非常に単純な単語カウントを標準の Hadoop ジョブとして実行しています。

この出力エラーは、一時出力用の一時ディレクトリを作成できないことを示しています。

MRJob でデフォルトで必要なディレクトリを作成できないとします。コマンドラインから MRJob にオプションを渡すことはできますか? これまでに見つけた唯一のオプションはbase_tmp_dirです。説明には、「ローカル一時ディレクトリを配置するパス」と記載されていました。一時出力ディレクトリはhdfsにあるはずなので、「ローカル」は正確には探しているものではありません。それでも、やってみるつもりでした(:

また

しかし、MRJob がそのようなオプションがないと文句を言うので失敗しました:

は、ここword_count.pyにある標準的なものです。MRJobj に関する重要な知識が欠けているか、Hadoop ストリーミングに戻らなければならない可能性があります。