如何使用pyspark从spark dataframe中的Column中读取JSON作为字符串或字典，而不使用RDD或collect()？

b4qexyjb 于 2023-05-19 发布在 Spark

关注(0)|答案(2)|浏览(198)

我有一个像这样的数据框

+-------------------------------------------+
|                                     output|
+-------------------------------------------+
|{"COLUMN1": "123", "COUMN2": {"A":1 "B":2}}|
+-------------------------------------------+

我只是想把JSON作为一个字符串或字典读到变量中，这样我就可以对它做进一步的操作。
问题是-

显然，当你在数据块上使用unity catalogue时，你不允许使用rdd或方法，如.iterrows，.collect等。（ref - https://community.databricks.com/s/question/0D58Y00009yKdeHSAS/cannot-use-rdd-and-cannot-set-sparkdatabrickspysparkenablepy4jsecurity-false-for-cluster）
使用.asDict或.first（）等方法转换为Rows数据类型，无法将其转换回json。例如。

Row(output=Row(COLUMN1='123', ...

DF是如何创建的？

nextdf = df.select(struct(col("COLUMN1"),col("COLUMN2"),col("COLUMN3")).alias("output"))

输出应为-{"COLUMN1": "123", "COUMN2": {"A":1 "B":2}}
请让我知道我可以尝试什么？

JSON

来源：https://stackoverflow.com/questions/76280090/how-do-i-read-a-json-as-a-string-or-dictionary-from-a-column-in-spark-dataframe

2条答案

按热度按时间

dtcbnfnu1#

这种情况下可以使用**toJSON()**。

Example:

from pyspark.sql.functions import *
df = spark.createDataFrame([(1, "foo"),(2, "bar"),],["id", "label"])

df1= df.withColumn("temp", concat_ws(" ", *df.columns)).groupBy(lit(1)).agg(array_join(collect_list(col("temp"))," ").alias("new_column")).drop("1")

print(df1.select(struct(col("new_column")).alias("new")).toJSON().collect()[0])
#{"new":{"new_column":"1 foo 2 bar"}}

得到东西而不收集然后使用

Save as Text:

使用**save作为text文件，带有header**标志false，从输出文件中转义列名。

df.coalesce(1).write.format("text").option("header", "false").save("output.txt")

赞(0）回复(0）举报 2023-05-19

7hiiyaii2#

RDD是旧接口，DataFrame是新/替换接口。使用DataFrame方法。

my_var = nextdf.collect()[0].asDict()['output']
print(my_var)
# should print `{"COLUMN1": "123", "COUMN2": {"A":1 "B":2}}`

赞(0）回复(0）举报 2023-05-19

我来回答

如何使用pyspark从spark dataframe中的Column中读取JSON作为字符串或字典，而不使用RDD或collect()？

2条答案

相关问题

热门标签

最新问答