50 台のマシンで Spark クラスターを実行しています。各マシンは 8 コアの VM で、メモリは 50 GB です (41 は Spark で使用できるようです)。

私はいくつかの入力フォルダーで実行しています。入力のサイズは gz 圧縮で ~250GB と見積もっています。

私が使用しているマシンの数と構成は十分であるように思えますが、約 40 分の実行後にジョブが失敗し、ログに次のエラーが表示されます。

また:

このような問題をデバッグするにはどうすればよいですか?

編集：問題の根本原因を見つけました。それはこのコードです：

特に、BufferedInputStream を使用してファイルのコンテンツを読み取るために、各ファイルにサイズ 40M のバッファーを割り当てます。これにより、スタック メモリがある時点で終了します。

問題は次のとおりです。

• 行ごとに読み取ると（バッファは必要ありません）、非常に非効率的な読み取りになります
• 1 つのバッファーを割り当て、ファイルの読み取りごとに再利用すると、並列処理の意味で可能ですか? それとも、いくつかのスレッドによって上書きされますか?

どんな提案も大歓迎です...

編集 2: バイト配列の割り当てをイテレータの外に移動することで最初のメモリの問題を修正し、すべてのパーティション要素で再利用されるようにしました。ただし、分割の目的で作成される新しい String(buffer, "UTF-8").trim()) がまだあります。これは、毎回作成されるオブジェクトでもあります。stringbuffer/builder を使用できますが、String オブジェクトなしで文字セット エンコーディングを設定するにはどうすればよいでしょうか。

80 台のマシンで Spark クラスターを実行しています。各マシンは 8 コアの VM で、メモリは 50 GB です (41 は Spark で使用できるようです)。

私はいくつかの入力フォルダーで実行しています。入力のサイズは gz 圧縮で ~250GB と見積もっています。

ドライバ ログにエラーが表示されます。何を確認すればよいかわかりません。例 (ログに表示される順序):

私のジョブ クラス (Phase0) はどのスタック トレースにも含まれていないため、これらのエラーから問題の原因について何を学べるかわかりません。助言がありますか？

編集:具体的には、数GBのフォルダーで作業している場合でも、次の例外が発生します:

「Launch click-to-deploy software」機能を使用して、Google Compute Engine で Hadoop クラスターをセットアップしようとしています。1 つのマスター ノードと 1 つのスレーブ ノードを作成し、マスター ノードから start-all.sh スクリプトを使用してクラスターを開始しようとしました。 「許可が拒否されました（公開鍵）」というエラーが発生しました。

スレーブ ノードとマスター ノードの両方で公開鍵と秘密鍵を生成しました。

現在、自分のユーザー名でマスターにログインしていますが、「hadoop」ユーザーとしてマスターにログインすることは必須ですか?もしそうなら、そのユーザー ID のパスワードは何ですか?

この問題を克服する方法を教えてください。

Spark 1.1 の使用

私は次のような仕事をしています：

1. 特定のルートの下にあるフォルダーのリストを読み取り、リストを並列化します
2. 各フォルダについて、その下のファイルを読み取ります - これらは gzip されたファイルです
3. ファイルごとにコンテンツを抽出します。これらは行であり、各行は単一のイベントを表し、フィールドはタブ (TSV) で区切られています
4. すべての行の単一の RDD を作成します。
5. TSV を json に変換します。

(現在、線は特定のイベント タイプを表しています。4 つのタイプがあります: セッション、リクエスト、レコメンデーション、ユーザー イベント)

6. セッション イベントのみを除外します。一部のユーザーIDフィールドに従って、それらの1:100のみをサンプリングします。いくつかの出力構造 (イベントの種類/日付/イベントなど) を表すキーを使用して、それらをペアに変換し、FS に書き込みます。
7. リクエストとユーザー イベントについても同じことを行います

(レコメンデーションの場合、ユーザー ID に従ってサンプリングを行うことはできません (ユーザー ID がそこに存在しないため)。ただし、相互のリクエスト ID フィールドに基づいて、リクエストとレコメンデーションの間に 1 対 1 の関係があることがわかっています。したがって、:)

8. 個別のリクエスト ID のリストを作成します。このリストをリクエスト ID に基づく推奨リストにキーとして結合し、必要なフィルタリングを実現します。次に、縮小されたリストを FS に出力します。

さて、ここに私の問題があります。これらのことを行うために使用するコードは、小規模で機能します。しかし、比較的大きな入力で実行し、8 コアとそれぞれ 50 GB のメモリを備えた 80 台のマシンのクラスターを使用すると、多くのマシンが使用されていないことがわかります。つまり、1 つのコアしか占有されていません (また、約 20% しか使用されていません)。メモリは、ジョブに構成された 40 GB のうち 16 GB のみです。

どこかで変換がうまく並列化されていないと思いますが、どこで、なぜなのかわかりません。これが私のコードのほとんどです（問題とは無関係だと思われる補助機能の一部を省略しています）

最後に、Spark ジョブを開始するために使用するコマンドは次のとおりです。

クラウド ストレージに Hadoop コネクタを使用する予定がある場合でも、Google クラウドの Click to deploy Hadoop ワークフローでローカル永続ディスクのサイズを選択する必要があるのはなぜですか? デフォルトのサイズは 500 GB です。ディスクが必要な場合は、サイズをもっと小さくする必要があると考えていました。Google クラウドの Hadoop でクラウド ストレージ コネクタを使用する場合に推奨される永続ディスク サイズはありますか?

「Google Cloud Platform に Apache Hadoop をデプロイする

Apache Hadoop フレームワークは、コンピューターのクラスター全体にわたる大規模なデータ セットの分散処理をサポートします。

Hadoop は単一のクラスターにデプロイされます。デフォルトのデプロイでは、1 つのマスター VM インスタンスと 2 つのワーカー VM が作成され、それぞれに 4 つの vCPU、15 GB のメモリ、および 500 GB のディスクがあります。クラスターのセットアップを管理するために、一時的なデプロイ コーディネーター VM インスタンスが作成されます。

Hadoop クラスタは、Cloud Storage バケットをデフォルトのファイル システムとして使用し、Google Cloud Storage コネクタを介してアクセスします。Cloud Storage ブラウザにアクセスして、Hadoop デプロイで使用できるバケットを検索または作成します。

Google Compute Engine 上の Apache Hadoop クリックしてデプロイ Apache Hadoop Apache Hadoop ZONE us-central1-a WORKER NODE COUNT

CLOUD STORAGE BUCKET バケットを選択 HADOOP VERSION 1.2.1 MASTER NODE DISK TYPE Standard Persistent Disk MASTER NODE DISK SIZE (GB)

WORKER NODE DISK TYPE Standard Persistent Disk WORKER NODE DISK SIZE (GB) "

Flume-ng から Google Cloud Storage にデータを書き込みたいと考えています。非常に奇妙な動作を観察したため、少し複雑です。説明させてください：

序章

バケットを使用するように設定された Google クラウド (ワンクリック) で Hadoop クラスターを起動しました。

マスターで ssh し、hdfsコマンドを使用してファイルを追加すると、バケットですぐに確認できます

しかし、コンピューターから追加して読み取ろうとすると、他の HDFS を使用しているようです。ここに というファイルを追加しましたjp.txtが、以前のファイルが表示されませんtest.txt

これは、HDFS を調べたときに表示される唯一のファイルでもありますhttp://ip.to.my.cluster:50070/explorer.html#/

Web コンソール ( https://console.developers.google.com/project/my-project-id/storage/my-bucket/ )を使用してバケット内のファイルを一覧表示すると、のみが表示され、 は表示test.txtされませんjp.txt。

Hadoop が Google Cloud Storage に接続できないことを読み、それに応じて Hadoop クライアントを構成した (かなり難しいことです) と、バケット内のアイテムが表示されるようになりました。しかし、そのためにはgs://URIを使用する必要があります

観察・中間結論

したがって、同じクラスタに 2 つの異なるストレージ エンジンがあるようです: 「従来の HDFS」( で始まるhdfs://) と Google ストレージ バケット ( で始まるgs://)。

ユーザーと権限は、ファイルをどこからリストするかによって異なります。

質問

主な質問は次のとおりです。flume を使用して Google Cloud Storage 上の HDFS/GS に書き込むために必要な最小限の設定は何ですか?

関連する質問

• 目標を達成するために、Google Cloud で Hadoop クラスタを起動する必要がありますか?
• Google Cloud Storage Bucket に直接書き込むことはできますか? はいの場合、どのようにflumeを構成できますか? (jar の追加、クラスパスの再定義...)
• 同じクラスターに 2 つのストレージ エンジンがあるのはなぜですか (従来の HDFS / GS バケット)

私のフルーム構成

行 a1.sinks.hdfs_sink.hdfs.path はgs://パスを受け入れますか?

その場合、どのようなセットアップが必要ですか (追加の jar、クラスパス) ?

ありがとう

Google Compute Engine で Hadoop クラスタをセットアップしようとしており、こちらの手順に従っています。私が走るまで、すべてがうまくいったようです：

プロジェクト ID と作成したバケット名を使用します。スクリプトは何かにアクセスできないようで、403 でクラッシュします。エラーメッセージを含む出力の末尾は次のとおりです。