問題タブ [pyspark]

forEach に関する多くのドキュメントが見つからないようです。キーと値のペアになっているデータセットがあります。私は（疑似コード）のようなことをしようとしています：

forEach キー、forEach キーの値の合計、値の最大値など。

だから私がやったことは

データのキーを 0 ～ 49 にします。それから私はすることにしました：

奇妙なことが起こっていることに気付きました。次のファイル サイズでは、計算に 10 GB で 46 秒、50 GB ファイルで 10 分 31 秒かかりました。ファイルを確認したところ、何らかの理由で 4 ブロックしかありませんでした。

だから私がしたことは変更されました：

でパーティションを削除したところ、50GB のファイルが約 1 分に減少しました。ロード後にデータを再パーティション化することはまだ理にかなっているのだろうか? もしかしてキーで？

次の単純な wordcount Python スクリプトがあります。

このコマンド ラインを使用して、このスクリプトを起動しています。

次のエラーが表示されます。

私のような Spark 初心者には、これが問題のようです:「エラー シェル: Hadoop バイナリ パスで winutils バイナリを見つけられませんでした」。ただし、Spark のドキュメントには、Spark をスタンドアロン モードで実行するために Hadoop をインストールする必要はないと明確に記載されています。

私は何を間違っていますか？

結合を実行していますが、データが 100 を超えるノードにまたがっています。したがって、別のキーと値のペアと結合するキーと値の小さなリストがあります。

私のリストは次のようになります。

私はブロードキャスト変数を持っています:

私が参加するとき：

次のエラーが表示されます。

たとえば、ネストされた辞書のリストがあり、ネストされた辞書のスキーマを推測ds = [{'a': {'b': {'c': 1}}}]しながら、そこからspark DataFrameを作成したいと考えています。を使用すると、次のスキーマが得られますsqlContext.createDataFrame(ds).printSchema()

しかし、私が必要なのはこれです

2 番目のスキーマは、最初に辞書を JSON に変換してから、次のjsonRDDようにロードすることで作成できますsqlContext.jsonRDD(sc.parallelize([json.dumps(ds[0])])).printSchema()。ただし、これは大きなファイルの場合は非常に面倒です。

pyspark.sql.Row()データフレームがスキーマを推測することを期待して、辞書をオブジェクトに変換することを考えましたが、辞書に異なるスキーマがある場合 (たとえば、最初にいくつかのキーが欠落していた場合) は機能しませんでした。

これを行う他の方法はありますか？ありがとう！

pyspark (Apache Spark) の DataFrame API を使用していますが、次の問題が発生しています。

同じソース DataFrame に由来する 2 つの DataFrame を結合すると、結果の DF は膨大な数の行に爆発します。簡単な例:

nディスクから行を含む DataFrame をロードします。

次に、そのソースから 2 つの DataFrame を作成します。

最後に、(内部で) それらを元に戻したい:

キーインcol1はユニークです。n結果の DataFrame には行があるはずですが、行がありn*nます。

ディスクから直接ロードするdf_oneと、それは起こりません。df_two私は Spark 1.3.0 を使用していますが、これは現在の 1.4.0 スナップショットでも発生します。

なぜそれが起こるのか誰か説明できますか？

map reduce でキーを分割し、新しいキーと値のペアを作成したいと考えています。

現在のドキュメント ファイル:

次のように各値でキーを分割したい：

どんな助けでも大歓迎です！ありがとう

私はいくつかの Spark アプリケーションを開発するために Python コードを書いています。Python が実行中の JVM とどのようにやり取りするのか、非常に興味があり、Spark のソース コードを読み始めました。

最終的に、すべての Spark 変換/アクションが、次の方法で特定の jvm メソッドを呼び出していることがわかります。

_jvmPython プログラマーとして、このオブジェクトで何が起こっているのか非常に興味があります。ただし、 pysparkの下のすべてのソース コードを簡単に読んだところ、 Context_jvmクラスの属性であることがわかりました。それ以上は、属性やメソッドについては何も知りません。_jvm's

pyspark が JVM 操作にどのように変換されるかを理解してくれる人はいますか? scala コードを読んで、_jvmそこに が定義されているかどうかを確認する必要がありますか?

高次元の寄木細工のファイルを読み込んでいますが、数列しか必要ありません。私の現在のコードは次のようになります。

何が起こっているかについての私のメンタル モデルは、すべてのデータをロードしてから、不要な列を捨てているというものです。私は明らかに、それらのコラムを読まないことを望んでおり、寄木細工について理解していることから、それは可能であると思われます。

したがって、2 つの質問があります。

1. 私のメンタルモデルは間違っていますか？または、spark コンパイラは、上記の例の列 a、b、および c のみを読み取るほどスマートですか?
2. sqc.parquetFile()データをより効率的に読み込むにはどうすればよいですか?

python-snappy を使用してファイルを圧縮し、hdfs ストアに配置しました。私は今、そのように読み込もうとしていますが、次のトレースバックが表示されます。ファイルを読み込む方法の例が見つからないので、処理できます。テキストファイル（非圧縮）版は問題なく読めます。sc.sequenceFile を使用する必要がありますか? ありがとう！

() 内の ValueError トレースバック (最新の呼び出しが最後) ----> 1 a_file.first()

/home/user/Software/spark-1.3.0-bin-hadoop2.4/python/pyspark/rdd.pyc in first(self) 1244 if rs: 1245 return rs[0] -> 1246 raise ValueError("RDD is空") 1247 1248 def isEmpty(自己):

ValueError: RDD が空です

出力: u'##fileformat=VCFv4.1'