Pyspark 从具有不同列的行/数据创建 DataFrame

Question

我有多个键/值对的数据/行，键的数量未知——有些重叠，有些没有——我想从中创建一个 Spark DataFrame。我的最终目标是从此 DataFrame 编写 CSV。

我对输入数据/行具有灵活性：它们最容易是 JSON 字符串，但可以通过 可能重叠的键进行转换：

{"color":"red", "animal":"fish"}
{"color":"green", "animal":"panda"}
{"color":"red", "animal":"panda", "fruit":"watermelon"}
{"animal":"aardvark"}
{"color":"blue", "fruit":"apple"}

理想情况下，我想根据这些数据创建一个如下所示的 DataFrame：

-----------------------------
color | animal   | fruit
-----------------------------
red   | fish     | null
green | panda    | null
red   | panda    | watermelon
null  | aardvark | null
blue  | null     | apple
-----------------------------

值得注意的是，没有特定键的数据/行是null，数据/行中的所有键都表示为列。

我对 Spark 的许多基础知识感到相对自在，但在设想一个有效地使用键/值对获取我的 RDD/DataFrame 的过程时遇到了困难 -- 但列和键的数量未知 -- 并创建一个以这些键为列的 DataFrame。

高效，因为如果可能，我想避免创建一个所有输入行都保存在内存中的对象（例如单个字典）。

再一次，编写 CSV 的最终目标是，我假设创建一个 DataFrame 是实现这一目标的合乎逻辑的步骤。

另一个皱纹：

一些数据将是多值的，例如：

{"color":"pink", "animal":["fish","mustang"]}
{"color":["orange","purple"], "animal":"panda"}

使用提供的分隔符，例如/ 为避免与 , 冲突以分隔列，我想在列的输出中分隔这些，例如：

------------------------------------
color         | animal       | fruit
------------------------------------
pink          | fish/mustang | null
orange/purple | panda        | null
------------------------------------

一旦有了解决主要问题的方法，我相信我可以解决这部分问题，但无论如何都会把它扔掉，因为这将是问题的一个维度。

Answer 1

从文件中读取

如果您的数据存储在一个文件中（假设它被命名为myfile.json），如下所示：

{"color":"red", "animal":"fish"}
{"color":"green", "animal":"panda"}
{"color":"red", "animal":"panda", "fruit":"watermelon"}
{"animal":"aardvark"}
{"color":"blue", "fruit":"apple"}
{"color":"pink", "animal":["fish","mustang"]}
{"color":["orange","purple"], "animal":"panda"}

您可以使用pyspark.sql.DataFrameReader.json 将文件读取为换行符分隔的 JSON 记录。

df = spark.read.json("myfile.json")
df.show()
#+------------------+-------------------+----------+
#|            animal|              color|     fruit|
#+------------------+-------------------+----------+
#|              fish|                red|      null|
#|             panda|              green|      null|
#|             panda|                red|watermelon|
#|          aardvark|               null|      null|
#|              null|               blue|     apple|
#|["fish","mustang"]|               pink|      null|
#|             panda|["orange","purple"]|      null|
#+------------------+-------------------+----------+

df.printSchema()
#root
# |-- animal: string (nullable = true)
# |-- color: string (nullable = true)
# |-- fruit: string (nullable = true)

从 RDD 读取

您也可以这样做来读取rdd：

import json

rdd = sc.parallelize(
    map(
        json.dumps,
        [
            {"color":"red", "animal":"fish"},
            {"color":"green", "animal":"panda"},
            {"color":"red", "animal":"panda", "fruit":"watermelon"},
            {"animal":"aardvark"},
            {"color":"blue", "fruit":"apple"},
            {"color":"pink", "animal":["fish","mustang"]},
            {"color":["orange","purple"], "animal":"panda"}
        ]
    )
)

df = spark.read.json(rdd)

---

对于第二部分，您可以根据需要使用pyspark.sql.functions.regexp_replace 来格式化您的多值记录。

from pyspark.sql.functions import regexp_replace

def format_column(column):
    return regexp_replace(regexp_replace(column, '(^\[)|(\]$)|(")', ''), ",", "/") 

df.select(*[format_column(c).alias(c) for c in df.columns]).show()

#+------------+-------------+----------+
#|      animal|        color|     fruit|
#+------------+-------------+----------+
#|        fish|          red|      null|
#|       panda|        green|      null|
#|       panda|          red|watermelon|
#|    aardvark|         null|      null|
#|        null|         blue|     apple|
#|fish/mustang|         pink|      null|
#|       panda|orange/purple|      null|
#+------------+-------------+----------+