pyspark

5熱

1答えて

内部計算にJavaオブジェクトを使用するpyspark pythonで使用するUDFを作成する必要があります。それは私のようなものだろう、単純なパイソンた場合： def f(x): return 7 fudf = pyspark.sql.functions.udf(f,pyspark.sql.types.IntegerType()) をして使用してそれを呼び出す： df = sq

0熱

1答えて

pPySpark行オブジェクト：変数名で行要素にアクセスする

を指定すると、r.nameまたはr.ageをそれぞれ使用して、ドット表記を使用してPySpark 要素にアクセスできます。名前が変数elementに保存されている要素を取得する必要がある場合はどうなりますか？ 1つの方法はr.toDict()[element]です。しかし、大きな数字のDataFrameがあり、そのデータフレームの各行に関数をマップしたい場合を考えてみましょう。我々は確かしかし、す

2熱

1答えて

スパークジョブのログ

私はクラスタ上で送信されているジョブを追跡しようとしていますが、spark.eventLog.enabled = Trueを使用してイベントログのログソリューションを見つけました。フィニッシュ（詳細はhere）またはlog4jはタスクの状態と進行状況に関する情報も提供します。私が本当に望むのは、実行されているタスクを記録することです。これは、var = sc.range（1000）またはmin_

2熱

2答えて

は、私はこのコードを持っているPyspark

でシンプルな行にデータフレームの複雑な行を分割： from pyspark import SparkContext from pyspark.sql import SQLContext, Row sc = SparkContext() sqlContext = SQLContext(sc) documents = sqlContext.createDataFrame([ Row

2熱

1答えて

pyspark行列アキュムレータ

pyspark accumulatorを使用してrddから推測された値を行列に追加したいです。私は文書が少し不明であることを発見した。バックグラウンドのビットを追加します。私のrddDataには、1つのカウントをマトリックスに追加しなければならないインデックスのリストが含まれています。たとえば、このリストはインデックスにマップ： from pyspark.accumulators import

2熱

1答えて

numpy行列をpyspark rddに変換する

私は2dのnumpyの配列を持っています。行列の各行がrddのエントリであるところからpyspark rddを作成するにはどうすればよいですか？ように： rddData.take(1)[0] == list(aaData[0]) aaDataがnumpyの2Dアレイ（行列）であり、rddDataはaaDataから作成RDDありますか？

3熱

2答えて

PySpark + Google Cloud Storage（wholeTextFiles）

私はPySpark（Google Dataproc）を使用して約100万のHTMLファイルを解析し、関連するフィールドを凝縮ファイルに書き出しようとしています。各HTMLファイルは約200KBです。したがって、すべてのデータは約200GBです。データのサブセットを使用すると、以下のコードは正常に動作しますが、数時間実行されてからデータセット全体を実行するとクラッシュします。さらに、ワーカーノード

0熱

1答えて

DStreamごとにRDDのすべての値の合計を見つけるにはどうすればよいですか？

私はスパークストリーミングを使用して、カフカからデータを連続的に読み込み、いくつかの統計を実行しています。私は毎秒ストリーミングしています。だから私は1秒バッチ（dstreams）を持っています。このdstream内の各RDDにはJSONが含まれています。私クリーン DSTREAMにRDDS OF kafkaStream = KafkaUtils.createDirectStream(strea

7熱

2答えて

列の値がsparkのリストと等しいかどうかによるフィルタリング

列の値がリストと等しいかどうかに基づいて、スパークデータフレームをフィルタリングしようとしています。 filtered_df = df.where(df.a == ['list','of' , 'stuff']) filtered_dfだけfiltered_df.aの値が['list','of' , 'stuff']あるとaのタイプはarray (nullable = true)ある行を含む場

1熱

1答えて

外部の__main__ファイルからのSparkContextをspark-submitに変更

私はpythonの依存関係をパッケージ化して、というハープ・クラスタに送信しようとしています。これはDRYestの方法で可能です。私は私のmy_spark_app.pyは次のようになりたい： from pyspark import SparkContext, SparkConf conf = SparkConf().setAppName('MyApp').setMaster('yarn-