問題タブ [spark-dataframe]

セットの分析をしようとしています。次のようなサンプル データ セットがあります。

注文.json

それは、ID を表す数字のリストである単一のフィールドだけです。

実行しようとしているSparkスクリプトは次のとおりです。

作成expandedしgroupedて問題ありません。一言で言えばexpanded、2 つの ID が同じ元のセットにあった、2 つの ID のすべての可能なセットのリストです。grouped自分自身と一致した ID を除外し、ID のすべての一意のペアをグループ化し、それぞれのカウントを生成します。のスキーマとデータ サンプルは次のgroupedとおりです。

だから、私の質問は、タプルのリストを得るために、各結果の最初の項目をグループ化するにはどうすればよいですか? 上記のサンプル データの場合、次のようなものが期待できます。

のスクリプトでわかるように、recs各行の最初のアイテムである「item1」で groupBy を実行することから始めると思いました。しかしその後は、この GroupedData オブジェクトに対するアクションが非常に制限されたままになります。実際には、sum、avg などの集計を行うだけです。各結果のタプルを一覧表示したいだけです。

この時点で RDD 関数を簡単に使用できましたが、それはデータフレームの使用から逸脱しています。データフレーム関数でこれを行う方法はありますか。

pyspark データフレームをグループ化し、特定の列の分散を計算したいと思います。平均の場合、これは非常に簡単で、次のように実行できます

ただし、差異については、関数サブモジュールに集計関数がないようです（これは非常に一般的な操作であるため、理由も疑問に思っています）

キーと値を含むテキスト ファイルがあり、一部の値が欠落していました。

欠損値を埋めたかったので、そうしました（データは私のrddです）

data1 は次のようになります。

上記のdata1をデータフレーム/テーブルに書き込みたい

しかし、私はエラーが発生しています

1)データフレームに書き込むにはどうすればよいですか? 2)以下のようにキーとリストを単一のタプルに変換するにはどうすればよいですか?

データフレームに直接書き込むことができるようにするには?

前もって感謝します ：）

csv ファイルがあります。私は pyspark で DataFrame(df) に変換します。いくつかの変換の後; df に列を追加したい。これは単純な行 ID (0 または 1 から N まで) である必要があります。

rdd で df を変換し、「zipwithindex」を使用します。結果のrddをdfに変換しました。このアプローチは機能しますが、250k のタスクが生成され、実行に多くの時間がかかります。実行時間を短縮する他の方法があるかどうか疑問に思っていました。

以下は私のコードのスニペットです。私が処理しているcsvファイルはBIGです。何十億もの行が含まれています。

私はDataFrame APIを使用しています。

既存の DataFrame と List オブジェクトがあります (Array も使用できます)。このリストを既存の DataFrame に新しい列として追加するにはどうすればよいですか? これにはクラスColumnを使用する必要がありますか?