問題タブ [pyspark]

Python メソッド内からブロードキャスト変数を作成しようとしています (分散操作に依存する、作成中のいくつかのユーティリティ メソッドを抽象化しようとしています)。ただし、Spark ワーカー内からブロードキャスト変数にアクセスできないようです。

この設定があるとしましょう：

ただし、代わりに仲介者を排除すると、SomeMethod()うまく機能します。

可能であれば、すべての Spark ロジックをメイン メソッドに配置する必要はありません。ローカル関数内から変数をブロードキャストし、それらを Spark ワーカーにグローバルに表示する方法はありますか?

別の方法として、この種の状況に適した設計パターンは何でしょうか?たとえば、自己完結型で、再利用したい特定の機能を実行する Spark 専用のメソッドを作成したいですか?

Spark のスタンドアロン モードで ec2-script を使用して 10 ノードのクラスターを起動しました。PySpark シェル内から s3 バケットのデータにアクセスしていますが、RDD で変換を実行すると、1 つのノードしか使用されません。たとえば、以下は CommonCorpus からデータを読み込みます。

これを実行すると、10 個のスレーブのうち 1 つだけがデータを処理します。これは、1 つのスレーブ (213) だけが、Spark Web コンソールから表示したときにアクティビティのログを持っているためです。Ganglia でアクティビティを表示すると、この同じノード (213) が、アクティビティの実行時にメモリ使用量が急増した唯一のスレーブです。

さらに、スレーブが 1 つだけの ec2 クラスターで同じスクリプトを実行すると、まったく同じパフォーマンスが得られます。私は Spark 1.1.0 を使用しています。ヘルプやアドバイスをいただければ幸いです。

Java のサードパーティ データベース クライアント ライブラリがいくつかあります。私はそれらにアクセスしたい

例: クライアント クラス (JDBC ドライバーではありません!) を Java ゲートウェイ経由で Python クライアントが利用できるようにするには:

サードパーティのライブラリを JVM クラスパスに追加する場所が明確ではありません。ファイルcompute-classpath.shに追加しようとしましたが、うまくいかなかったようです。私は得る：

Py4jError: パッケージを呼び出そうとしています

また、Hive と比較すると、ハイブ JAR ファイルはファイルcompute-classpath.shを介してロードされないため、疑わしいと思います。JVM 側のクラスパスをセットアップするための別のメカニズムが発生しているようです。

以下のコマンドを使用して、Spark 開発用の iPython ノートブックを作成しました。

そして、sc次のような Python コードを使用して SparkContext を作成しました。

spark.executor.memoryドキュメントの をよりよく理解したい

JVM メモリ文字列と同じ形式で、executor プロセスごとに使用するメモリの量

これは、1 つのノードで実行されているすべてのプロセスの累積メモリがその上限を超えないということですか? その場合、その数値をできるだけ高い数​​値に設定する必要がありますか?

これはいくつかのプロパティのリストでもあります。パフォーマンスを向上させるためにデフォルトから微調整できるパラメータが他にもあります。

ありがとう！

数千のファイルを入力として取り、Amazon S3 からダウンロードし、各マップ ステップが文字列を返すマップ フェーズで処理する Spark ジョブがあります。.tar.gz出力をファイルに圧縮し、後で S3 にアップロードしたいと思います。それを行う1つの方法は

問題は、outputsメモリに収まらないことです (ただし、ディスクには収まります)。マップフェーズで出力をマスターファイルシステムに保存する方法はありますか? それとも、ループfor output in outputsをジェネレーターとして使用して、すべてをメモリにロードする必要がないようにしますか?

Spark/Hadoop 入力言語である Pyspark: データセットで "SJC" などのキーワードを検索し、キーワード "SJC" が見つかった行に対応する 2 番目の列からテキストを返したいと考えています。

たとえば、次のデータセットは次のようになります。

[年] [遅延] [目的地] [フライト番号]

|1987| |-5| |SJC| |500|

|1987| |-5| |SJC| |250|

|1987| |07| |SFO| |700|

|1987| |09| |SJC| |350|

|1987| |-5| |SJC| |650|

「SJC」をクエリして、[Delay] 値をリストまたは文字列として返すことができるようにしたいと考えています。

私はここまで来ましたが、運がありません:

助けてくれてありがとう！

私はsparkが初めてで、csvファイルのいくつかのフィールドに基づいてdistinct().count()を作ろうとしています。

Csv 構造 (ヘッダーなし):

.csv をロードするには、次のように入力しました。

lines次に、期待どおりに返された 3の個別のカウント:

idしかし、 let sayとに基づいて個別のカウントを作成する方法がわかりませんcountry。