問題タブ [emr]

boto 2.8.0 を使用して、S3 に保存されている大きなログ ファイルに対して EMR ジョブフローを作成しています。私は Elastic Mapreduce に比較的慣れていないので、この問題からジョブフローを適切に処理する方法を感じています。

問題のログファイルは、ログ サーバーから出力された日付に対応するキーを使用して s3 に保存されます/2013/03/01/access.log。これらのファイルは非常に大きいです。私の mapreduce ジョブは、Apache Pig スクリプトを実行します。このスクリプトは、ログ ファイルに保存されている uri パスの一部を単純に調べ、ビジネス ロジックに対応する一般化されたカウントを出力します。

boto のクライアント コードは、cli の入力として日時を受け取り、PigStep必要なすべての日付のインスタンスを使用してジョブフローをスケジュールします。したがって、次のようなものを渡すとpython script.py 2013-02-01 2013-03-01、29 日分の datetime オブジェクトが繰り返され、s3 のそれぞれの入力キーでピッグステップが作成されます。from_dateこれは、結果として得られるジョブフローに、との間の timedelta の各日に 1 つずつ、非常に多くのステップが含まれる可能性があることを意味しますto_date。

私の問題は、私の EMR ジョブフローが非常に遅く、ほとんどばかげていることです。それは今一晩実行されていますが、その例セットの途中まではできていません. このような多くのジョブフローステップを作成しているのに何か問題がありますか? クライアント コードで前処理して各日付のステップを作成するのではなく、代わりにさまざまなキーの豚スクリプトを一般化する必要がありますか? これは、Elastic Mapreduce の最適化を探すのに実行可能な場所ですか? AWS cli ruby​​ クライアントに渡された 1 か月分の同等のデータに対する同様のジョブのelastic-mapreduce実行には、約 15 分かかりました (このジョブは、同じ豚のスクリプトによって強化されました)。

編集

言うまでもなく、ジョブはタイプ m1.small の 2 つのインスタンスに対してスケジュールされていましたが、それ自体が問題である可能性があります。

S3にはスナップ圧縮されたサーバーログがたくさんあり、ElasticMapReduceでストリーミングを使用してそれらを処理する必要があります。ストリーミングマッパースクリプトに送信する前に解凍できるように、ログが（HFSにプルされる前に）すでに圧縮されていることをAmazonとHadoopに伝えるにはどうすればよいですか？

私が見つけることができる唯一のドキュメントはここにあります：http：//docs.aws.amazon.com/ElasticMapReduce/latest/DeveloperGuide/HadoopDataCompression.html#emr-using-snappy 、そしてそれは中間圧縮を参照しているようであり、 HFSに到着すると圧縮されます。

ところで、私は主にPythonで作業しているので、botoで解決策がある場合はボーナスポイントがあります！

Hive テーブルの列の 1 つに、キーと値のペアを格納したいと考えています。Hive の複雑なデータ型マップは、その構造をサポートしています。

(これは、私ができるようにしたいことのおもちゃの例にすぎません。このように圧縮したい列が他にもたくさんあります)

だから私はこのようなテーブルを作成します：

次に、コードで説明されているように、リデューサーを使用してマップ列を読み込もうとします: time_map の構造は次のようになります: {"min": time, "max": time, "average": time, "total":時間}

しかし、次のようなエラーが発生します。

Python reducer を使用してマップ列にロードするにはどうすればよいですか?

EMR (Elastic MapReduce) の後処理の例はありますか? 私が達成しようとしているのは、Amazon の Hadoop が仕事を終えた直後に、人々のグループに電子メールを送信することです。

S3 から HDFS に多数の小さな (200 ～ 600 KB) ファイルをマージするために、s3distcp を実行しようとしています。

Ubuntu を介して CDH 4.2 で Hadoop を実行しています。

具体的には: Hadoop 2.0.0-cdh4.2.0 Subversion file:///var/lib/jenkins/workspace/generic-package-ubuntu64-12-04/CDH4.2.0-Packaging-Hadoop-2013-02-15_10- 38-54/hadoop-2.0.0+922-1.cdh4.2.0.p0.12~正確/src/hadoop-common-project/hadoop-common -r 8bce4bd28a464e0a92950c50ba01a9deb1d85686

以前に、aws-java-sdk-1.4.1.jar および s3distcp.jar へのすべての依存関係を Hadoop クラスパスにコピーすることで解決しました。Libsnappy1 もインストールされます。

しかし、私が実行すると：

次のエラー スタックが表示されます。

他に試してみるべきことがありますか？正規表現で見えない問題はありますか?

編集：

namenode のログを見ると、定期的に例外が発生していることに気付きました。それは関連性がありますか？

map と reduce を行うカスタム バイナリを作成し、常識的な "cat file | map | sort | reduce > output" パターンを使用して正しい動作をテストしました。バイナリを静的にコンパイルして、可能な限り多くの依存関係を取り込むようにしました。また、バイナリをマスターに手動でアップロードすることにより、バイナリが Amazon の EMR AMI で実行されることも確認しました。関連する場合、選択した言語は Haskell であり、コンパイル結果は単純なネイティブのバイナリ実行可能ファイルになります。

最も単純なケースを取り上げます。

ジョブは開始されますが、マップ 0% フェーズでスタックし、動きません。そこから先に進むことはなく、どのログも有用なものを示していないようです。各マップ タスクは、600 秒で「レポートが返されない」ため強制終了されます。各マッパーは、0% の完了を示しながら、そのステータスとして次のようなものを示します。

カウンタ セクションには、s3n から読み取った 17.5KB のようなものが表示されます。

テストするためにジョブを次のように変更するとします。

その後、マッパー フェーズは 100% 終了しますが、リデューサーは次の例外を発生させます。

イライラする部分は、たとえば、同じ種類の EMR クラスターでハイブを実行しても、S3 で新しい外部マップ テーブル、したがってファイルを作成する際に問題がないように見えることです。

いくつかのアイデアを試してみましたが、セットアップを機能させるための正しい方向性を教えていただければ幸いです。

ありがとう、OA