come aggregare più colonne e generare una stringa formattata in Python

Question

Ho tabella / dateframe con la seguente struttura

**userid   item    createdon**
user1    item1   2020-10-01
user1    item2   2020-10-02
user1    item3   2020-10-03
user2    item4   2020-01-01
user2    item1   2020-03-03
...

per ogni userid, ho bisogno di generare una stringa in un formato come questo:

per utente1:

{ "date": "2020-10-01",
  "item": {"display": "item1"}
},
{ "date": "2020-10-02",
  "item": {"display": "item2"}
},
{ "date": "2020-10-03",
  "item": {"display": "item3"}
}

per utente2:

{ "date": "2020-01-01",
  "item": {"display": "item4"}
},
{ "date": "2020-03-03",
  "item": {"display": "item1"}
}

Sto usando pyspark. Mi chiedo se potrei ottenere questo risultato utilizzando la trasformazione della mappa? Qualsiasi aiuto sarebbe apprezzato.

Answer 1

Potrei far funzionare l'esempio. Scrive un file per utente.

Rif: Spark: scrive diversi file JSON da DataFrame in base alla separazione in base al valore della colonna

from pyspark.sql import Row
from pyspark.sql.types import *
from pyspark import SparkContext, SQLContext
from pyspark.sql.functions import *

sc = SparkContext('local')
sqlContext = SQLContext(sc)

data1 = [
        ("user1", "item1", "2020-10-01"),
        ("user1", "item2", "2020-10-02"),
        ("user1", "item3", "2020-10-03"),
        ("user2", "item4", "2020-01-01"),
        ("user2", "item1", "2020-03-03")
      ]

df1Columns = ["userid", "item", "createdon"]
df1 = sqlContext.createDataFrame(data=data1, schema = df1Columns)
df1.printSchema()
df1.show(truncate=False)

partialSchema = StructType([StructField("date", StringType(), nullable=True)
                           , StructField("item", StructType([StructField("display", StringType(), nullable=True)]), nullable=True)
                        ])

actualSchema = StructType([StructField("userid", StringType(), nullable=True)
                           , StructField("dict", partialSchema, nullable=True)
                        ])

res_df = df1.rdd.map(lambda row: Row(row[0], {"date": row[2], "item" : {'display':row[1]}}))\
    .toDF(actualSchema)

res_df.show(20, False)

res_df.repartition(col("userid")).select(col("userid"), col("dict.*")).write.partitionBy("userid").json("./helloworld/data/")

L'ultima riga scrive due file uno per ogni utente.

Contenuto del primo file utente:

{"date":"2020-10-01","item":{"display":"item1"}}
{"date":"2020-10-02","item":{"display":"item2"}}
{"date":"2020-10-03","item":{"display":"item3"}}

Contenuto del secondo file utente:

{"date":"2020-01-01","item":{"display":"item4"}}
{"date":"2020-03-03","item":{"display":"item1"}}