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

我有多个键/值对的数据/行，其中键的数量未知——有些重叠，有些不重叠——我想从中创建一个 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 
------------------------------------ 

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

请您参考如下方法:

从文件中读取

如果您的数据存储在如下文件中(假设它被命名为 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| 
#+------------+-------------+----------+