問題タブ [pyspark]

iPython Notebook を介して、ローカル マシンで Apache Spark の PySpark を使用してコードのプロトタイプを作成しています。正常に動作するように見えるコードをいくつか書きましたが、簡単な変更を加えると壊れてしまいます。

以下の最初のコード ブロックが機能します。2 番目のブロックは、指定されたエラーで失敗します。どんな助けにも本当に感謝します。このエラーは、Python オブジェクトのシリアル化に関係していると思われます。エラーは、TestClass を Pickle できないことを示しています。クラスをピクル可能にする方法に関する情報が見つかりません。ドキュメントには、「一般に、オブジェクトのすべての属性をピクルできる場合は、任意のオブジェクトをピクルできます。クラス、関数、およびメソッドをピクルすることはできません。オブジェクトをピクルすると、オブジェクトのクラスはピクルされず、何を識別する文字列にすぎません。これは、ほとんどのピクルスで問題なく機能します (ただし、ピクルスの長期保存に関する議論に注意してください)。TestClass を datetime クラスに置き換えてみたところ、問題なく動作しているように見えるため、これがわかりません。

とにかく、コード：

上記のコードは以下を返します。

347098

['テスト: こんにちは', 'テスト: こんにちは', 'テスト: こんにちは']

...終わり

しかし、「（1）この行を見る」とマークされた行（最後のマップ行）を削除して再実行すると、代わりに次のエラーが発生します。長いので、出力を投稿する前に、ここで質問を締めくくります。繰り返しますが、これについて助けていただければ幸いです。

前もって感謝します！

出力: 347098

からのこの出力には驚きましたfold。何をしているのか想像できません。

折り畳みは で始まり、要素ごとに追加されるため、something.fold(0, lambda a,b: a+1)の要素数が返されると思います。something01

私は Scala から来ました。そこでは、私が説明したように折り畳みが機能します。では、fold は pyspark でどのように機能するのでしょうか? ご意見ありがとうございます。

Spark でデータを処理していますが、1 日分のデータ (40G) で動作しますが、1 週間分のデータではOOMで失敗します。

異なる ID の数は 10k 未満です。各IDは小さめintです。OOM で失敗するエグゼキューターが多すぎるため、ジョブは失敗します。ジョブが (小さな入力で) 成功する"myoutput"と、約 100k になります。

1. 私は何を間違っていますか？
2. に置き換えsaveAsTextFileてみましたcollect（実際には、保存する前にPythonでスライスとダイシングを行いたいため）、動作に違いはなく、同じ失敗がありました。これは予想されることですか？
3. 私はreduce(lambda x,y: x.union(y), [sqc.parquetFile(...)...])代わりに持っていましたsc.union- どちらが良いですか? 違いはありますか？

クラスターには、 825 GB のRAM と224のコアを備えた25のノードがあります。

呼び出しはspark-submit --master yarn --num-executors 50 --executor-memory 5G.

1 つの RDD には最大 140 の列があり、1 時間のデータをカバーするため、1 週間は 168(=7*24) の RDD の結合になります。

オブジェクトを参照するクロージャを使用して RDD をフィルタリングすると、pickle エラーが発生します。

オブジェクトなし:

オブジェクトで：

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

テキストファイルを読み取り、メモリ内の登録済み一時テーブルとして使用するための作業コードがあります。スクリプトまたはモジュールのインポートを使用してこれらのテーブルのセットをロードし、対話的にクエリを実行したいと考えています。このコードをスクリプトと関数に入れたら、どちらのオブジェクトを返す必要がありますか? sc コンテキスト? テーブル？HadoopRDD?

ウィンドウ関数 ( などfirst, last, lag, lead) は でサポートされていpysparkますか?

たとえば、ある列でグループ化し、別の列で並べ替えてから、SparkSQL またはデータ フレームで各グループの最初の行を選択する (ウィンドウ関数のように) にはどうすればよいでしょうか?

pyspark.sql.functionsクラスに集計関数firstとが含まれていることがわかりlastましたが、クラスには使用できませんgroupBy。

私の問題は、 からsql.Rowとしてデータを読み込もうとしているときですString。私は pyspark を使用していますが、Scala API でもこの問題が発生していると聞きました。

pyspark.sql.Row オブジェクトはかなり非妥協的な生き物です。次の例外がスローされます。

つまり、フィールドの 1 つがバイト配列として表されています。次の python 印刷構造は機能しません

また

どちらも ClassCastException になります。

それで..他の人々はこれをどのように行うのですか？私は自分自身を転がし始めました（残念ながらここでコピー/貼り付けすることはできません..）しかし、これは少し車輪の再発明です..またはそう思います。

私は Spark-DataFrame API の初心者です。

このコードを使用して、タブ区切りのcsvをSpark Dataframeにロードします

新しいファイルから Spark で DataFrame を作成し、組み込みメソッド toPandas() を使用して pandas に変換するとします。

• Pandas オブジェクトをローカル メモリに保存しますか?
• Pandas の低レベルの計算はすべて Spark によって処理されますか?
• すべてのパンダデータフレーム機能を公開していますか? (私はそう思います)
• DataFrame APIにあまり触れずに、それをPandasに変換して、それで完了することはできますか?

PySpark でのいくつかの奇妙な動作に気づきました。洞察をいただければ幸いです。

単純な要素で構成されるRDDがあるとします

今、単純なクラスで、RDD のさまざまな属性をキャプチャすることに興味があります。たとえば、各要素から属性rdd.map(lambda s: getattr(s,'name'))を抽出するために使用します。name

したがって、このクラスのオブジェクト

それらを設定し、RDD からname対応するものを取得します。values

self.nameただし、これは式の中心にあると思われるエラーに遭遇しますlambda。この 2 番目のクラスは正常に動作します

ここで追加したのは、前の呼び出しだけで、代わりに にn=self.name渡さnれます。lambdaself.name

では、評価できない問題self.nameはlambdaですか? self.name純粋な pythonで同様の状況 ( を使用lambda) を作成しましたが、エラーはありません。これは Spark 固有のものだと思います。ご意見ありがとうございます。