問題タブ [apache-spark]

Spark 0.7.2 と Scala 2.9.3 でプログラミングを始めたばかりです。スタンドアロン マシンで機械学習アルゴリズムをテストしていますが、アルゴリズムの最後のステップでは、2 つのマトリックス間の MSE (平均二乗誤差) を計算する必要があります。A - M||^2 で、2 つの行列の間で要素ごとの減算を行います。A の潜在的なサイズは非常に大きくてまばらであるため、キーは座標 (i,j) であり、値は A の対応する要素とM、すなわち (A_ij, M_ij)。ML アルゴリズム全体が勾配降下であるため、反復ごとに MSE を計算し、特定のしきい値に対してテストします。ただし、プログラム全体は、反復ごとに MSE を計算せずに正常に実行されます。プログラムは次のようになります。

このプログラムは最大 45 回の繰り返ししか実行できず、次の Spark Exception でクラッシュします。

もう 1 つの観察結果は、反復ごとにランタイムが約 5% 増加することです。また、「reduce( _ + _ )」がないと、StackOverflowError は発生しません。可能な物理スレッドの総数まで並列処理を増やそうとしましたが、それは役に立ちません。

スタック オーバーフロー エラーの根本的な原因を特定できる方向性を誰かが指摘してくれたことを本当に感謝しています。

編集：

1. A の型は spark.RDD[((Double, Double), Array[Double])] です

2. stackoverflow 例外で、" at sun.reflect.GeneratedMethodAccessor4.invoke(Unknown Source)" から 61 回繰り返されます。

   /li>

3. 主な反復コード

次のリスト要素にはいくつかのユーティリティ関数が含まれています

以下を使用するいくつかのユーティリティ関数:

ローカル マシンで動作する単純な Spark-Streaming の例を取得しようとしています。
As/Bs/Cs をソケットに書き込むスレッドがあります。

As/Bs/Cs の数を数えようとする私のメイン プログラムは次のようになります (reduce ステップなし):

この場合、すべて正常に動作します (バッチのサンプル出力):

しかし、マップの後に削減ステップを追加すると、機能しなくなります。このコードは texts.print() の後に続きます

この場合、最初の「ストリーム」変数と「テキスト」変数の出力のみを取得し、reduce の出力は取得しません。また、この最初のバッチ処理の後は何も起こりません。また、spark ログ レベルを DEBUG に設定しましたが、例外やその他の奇妙なことは発生しませんでした。

そこで何が起こるの？ロックされるのはなぜですか?

Windows マシンで Spark クラスターをセットアップしようとしています。
ここに行く方法は、スタンドアロンモードを使用することですよね?
Mesos や YARN を使用しないことの具体的なデメリットは何ですか? そして、それらのいずれかを使用するのはどれほどの苦痛でしょうか? ここで経験した人はいますか？

Sparkシェルを実行しようとすると、これが得られます

参考までに、バインドできないと表示されている IP アドレスは、hadoop のマスター マシンの IP です (これを実行しているマシンとは異なります)。Hadoop と Scala の正しいバージョンを使用していますが、何が間違っているのかわかりません。どんな助けでも役に立ちます！:)

参加したいRDDが2つあり、次のようになります。

のキー値rdd1が一意であり、 のタプルキー値も一意である場合がありrdd2ます。次のrddが得られるように、2つのデータセットを結合したいと思います:

これを達成するための最も効率的な方法は何ですか? ここに私が考えたいくつかのアイデアがあります。

オプション1：

オプション 2:

オプション 1 では、マスタリングするすべてのデータが収集されますよね? そのため、rdd1 が大きい場合、これは適切なオプションとは思えません (私の場合は比較的大きいですが、rdd2 よりも 1 桁小さくなっています)。オプション 2 は醜く明確なデカルト積を行いますが、これも非常に非効率的です。私の頭をよぎった (まだ試していない) 別の可能性は、オプション 1 を実行してマップをブロードキャストすることですが、マップのキーがのキーrdd2。

誰もこのような状況に遭遇したことがありますか? お考えいただければ幸いです。

ありがとう！

Spark では、次のような Hadoop 構成設定を設定できます。

これは機能し、レプリケーション ファクターは 1 に設定されます。これが事実であると仮定すると、このパターン (通常の Hadoop 構成プロパティの前に "spark.hadoop." を追加する) は、textinputformat.record.delimiter に対しても機能すると考えました。

ただし、spark はこの設定を無視しているようです。textinputformat.record.delimiterを正しい方法で設定していますか? を設定する簡単な方法はありますかtextinputformat.record.delimiter。InputFormat2 つの改行で区切られたレコードを取得する必要があるだけなので、自分で を書くことは避けたいと思います。

公式の spark ドキュメントにはforeach、RDD で直接行われる呼び出しで使用されるアキュムレータの例があります。

独自のアキュムレータを実装しました:

ローカル設定では、これは問題なく機能します。ただし、複数のマシンを含む Spark スタンドアロン クラスターでこのジョブを実行すると、ワーカーは

アキュムレータをインクリメントする行でmyCounter。

私の質問は次のとおりです。アキュムレータは、ネストされた関数ではなく、RDD に直接適用される「トップレベル」の匿名関数でのみ使用できますか? はいの場合、呼び出しがローカルでは成功し、クラスターでは失敗するのはなぜですか?

edit : 例外の冗長性を高めました。

RDDは更新できますか? --- つまり、「a.txt」ファイルから RDD を作成したとします。「a.txt」を更新しました。ファイルa.txt全体を読み取らずにRDDを更新することは可能ですか?

EC2 の spark スクリプトによってセットアップされた EC2 クラスターで、distcp を使用して Hadoop と Amazon S3 の間でコピーしようとしていました。

私が得たエラーは

Spark でサンプルを実行しようとすると、エラーが発生します。Sparkでプログラムを実行するには、pom.xmlにどのような変更を加える必要があるか教えてください。