pyspark

1熱

1答えて

回答が見つかりません。私にRDDがある場合 rdd = sc.parallelize([('a', [1,2,3]), ('b',[4,5,6])]) 各値はリストです。今、各値はリストの要素の一つであり、キーと一致する場所、それが sc.parallelize([('a',1),('a',2),('a',3),('b',4),('b',5),('b'6)]) なるように、RDDを分割

0熱

1答えて

PySpark：DATAFRAME - 私は次のような構造でデータフレームを持つ配列

0熱

1答えて

同じパーティション：データフレームのために、後に、このデータフレーム

のRDDのために、私はデータフレームDFの仕様で複数のウィンドウ関数を使用します。 windowSpec = \ Window .partitionBy(df['customID']) 後、私が使用します。 df.rdd.map((x[0],x[0], x[1], x[2])) and df.mapPartitions(some func) 私のデータフレームのkv

-1熱

1答えて

TypeError：withColumn（）で複数の列を使用している場合、列は反復不可能です

日付の四半期の開始日を検索しようとしています。カラムは、私が使用しています df.withColumn("history_effective_quarter",add_months('history_effective_month',-(month('history_effective_month')%3)+1)) TypeError Traceback (most recent call l

2熱

1答えて

Pysparkの加重移動平均

私はPysparkの時系列の異常検出アルゴリズムを書いています。私は（-3,3）または（-4,4）ウィンドウの加重移動平均を計算したいと思います。今、私はラグを使い、ウィンドウ関数を導き、それらを重みの集合で掛けています。私のウィンドウは現在（-2,2）です。 Pysparkで加重移動平均を計算する別の方法があるかどうかを知りたいと思います。私が使用しています現在のコードは次のとおりです。 da

0熱

1答えて

postgresqlへの書き込みspark

MongoDBからApache SparkでPostgresに200万レコードを転送したい（両方のデータベースの公式ドライバを使用）。数分後にプログラムが停止し、私はpostgresログにこれらのエラーが見つかります： 2017-12-03T22:18:16+00:00 app[postgres.15348]: [AQUA] [7-1] sql_error_code = 26000 ERROR:

1熱

1答えて

複数のJSONオブジェクトを1行のファイルでPySparkに読み込むときのヌル値

S3バケットからPySpark DataFrameに一連のJSONファイルを取得する際に助けが必要です。このバケットのファイルはすべて、拡張子が.jsonですが、残念ながら通常のSparkでは1行に1つのJSONオブジェクトが必要ですが、大括弧で囲まれた1行になります。ので、代わりの： {"first_column": 12, "second_column": {"nested_column"

0熱

1答えて

Pysparkデータフレームのタイムスタンプ列に1時間を追加する

pysparkにはtest_timeという列があります。これはtimestamp列です。この列には、以下のようなレコードがあります。 2017-03-12 03:19:51.0 2017-03-12 03:29:51.0 今、私はtest_time列内のレコードに1 hourを追加したいです。結果： 2017-03-12 04:19:51.0 2017-03-12 04:29:51.0