PySpark私はPySparkを使用しています

Question

に大きなスパークデータフレーム内の行の各サブセットにマップの操作を行うと、私がやりたいことは以下のとおりです。方法：

は大きなスパークデータフレームDFは、すべてのレコードが含まれて

。私はこのdfの 'id'カラムで分割されたレコードの各サブセットで並列計算を行いたいと思っています。私は現在の考えることができる方法は次のとおりです。（私は説明するために単純な例を使用します）

dicts = [ 
    {'id': 1, 'name': 'a', 'score': 100}, 
    {'id': 1, 'name': 'b', 'score': 150}, 
    {'id': 2, 'name': 'c', 'score': 200}, 
    {'id': 2, 'name': 'd', 'score': 300}, 
] 
df = spark.createDataFrame(dicts) 

from pyspark.sql.functions import (
    collect_list, 
    struct 
) 

# df_agg will have the following schema: id, a list of structs 
df_agg = df.groupBy('id').agg(
    collect_list(struct(df.columns)).alias('records') 
) 

しかし、私は「my_func」は、いくつかの機能である

df_agg.rdd.map(my_func) 

をしよう主にSparkのデータフレーム計算を行い、いくつかの問題に直面し、どのように処理するか分からない。 my_funcは行上で動作し、行['records']は構造体のリストを保存します。この構造体のリストをSpark DataFrameに変換するにはどうしたらいいですか？

toDF（）は機能しません。 spark.createDataFrame（list、schema）を試しました。元のDFが使用するようにスキーマに入力しても、それは動作しません。

私はこれらのPySpark操作に比較的新しいので、このケースを処理する正しい方法が何であるか教えていただけたら助けてください。

ありがとうございます！

Answer 1

df_agg.rdd.map(my_func)（あなたがmy_funcの例を提供する場合は、私が行ってもよい）を試したときのエラーにコメントできません。しかし、あなたはDataFrameに変換できないと言いますので、ここにその部分の解決策を挙げます：

from pyspark.sql.types import StringType, StructField, StructType, BooleanType, ArrayType, IntegerType 

schema=StructType(
       [StructField("id", IntegerType(), True), \ 
       StructField("records", 
        ArrayType(StructType([StructField("id", IntegerType(), True),\ 
         StructField("name", StringType(), True),\ 
         StructField("score", IntegerType(), True)]))) 
       ]) 

df_agg.rdd.toDF(schema=schema).show(2)