Как преобразовать фрейм pyspark в JSON? - PullRequest
Новая
       25

Как преобразовать фрейм pyspark в JSON?

1 голос
/ 17 апреля 2020

У меня есть фрейм данных pyspark, и я хочу преобразовать его в список, содержащий JSON объект. Для этого я сделал, как показано ниже .. 

df.toJSON().collect()

Но эта операция отправляет данные драйверу, что является дорогостоящим и занимает много времени. И мой фрейм данных содержит миллионы записей. Так что есть еще один способ сделать это без операции collect (), которая оптимизирована, чем collect ().

Ниже приведен мой фрейм данных df: - 

      product cost
      pen      10
      book     40
      bottle   80
      glass    55

, и вывод такой, как показано ниже: - 

df2 = [{product:'pen',cost:40},{product:'book',cost:40},{product:'bottle',cost:80},{product:'glass',cost:55}]

, когда я распечатываю тип данных df2, это будет список.

1 Ответ

0 голосов
/ 17 апреля 2020

Если вы хотите создать json объект в кадре данных, используйте функции collect_list + create_map + to_json.

(or)

Чтобы записать как json документ в файл, не используйте to_json, вместо этого используйте .write.json()

Create JSON object:

df.agg(collect_list(create_map(lit("product"),"product",lit("cost"),"cost")).alias("stru")).\
selectExpr("to_json(stru) as json").\
show(10,False)

#+-------------------------------------------------------------------------------------------------------------------------------+
#|json                                                                                                                           |
#+-------------------------------------------------------------------------------------------------------------------------------+
#|[{"product":"pen","cost":"10"},{"product":"book","cost":"40"},{"product":"bottle","cost":"80"},{"product":"glass","cost":"55"}]|
#+-------------------------------------------------------------------------------------------------------------------------------+


#write to hdfs use .saveAsTextFile
df.agg(collect_list(create_map(lit("product"),"product",lit("cost"),"cost")).alias("stru")).selectExpr("to_json(stru) as json").rdd.map(lambda x:x['json']).saveAsTextFile("<path>")

#cat part-00000
#[{"product":"pen","cost":"10"},{"product":"book","cost":"40"},{"product":"bottle","cost":"80"},{"product":"glass","cost":"55"}]

Create JSON file:

df.agg(collect_list(create_map(lit("product"),"product",lit("cost"),"cost")).alias("stru")).write.mode("overwrite").json("<path>")

#cat part-00000-3a19165e-219e-4485-adb8-ef91589d6e31-c000.json
#{"stru":[{"product":"pen","cost":"10"},{"product":"book","cost":"40"},{"product":"bottle","cost":"80"},{"product":"glass","cost":"55"}]}
...