Pyspark：从列表的 RDD 创建一个火花数据框，其中列表的某些元素是对象

Posted 2023-04-15

【中文标题】Pyspark：从列表的 RDD 创建一个火花数据框，其中列表的某些元素是对象

【英文标题】：Pyspark: Create a spark dataframe from RDD of lists where some elements of list are objects

【发布时间】：2018-04-06 19:37:12

【问题描述】：

我正在尝试将 pandas.DataFrame 代码转换为等效的 pyspark DataFrame。 我有以下格式的 RDD。

myRdd = [[1, 'a', 'a':[1, 2]],
         [2, 'b', 'c': 1, 'd':3],
         [3, 'c', ]]

columnNames = ['sl', 'name', 'params']

内部列表中的第三个元素没有特定的结构。 在 pandas 数据框中，我可以将第三列视为 dtype=object。

pdDF = pandas.DataFrame(myRdd, columns=columnNames)

我可以做类似的事情来将上述格式的 pyspark RDD 转换为 pyspark DataFrame 吗？

【参考方案1】：

Spark 中没有dtype=object，但您可以定义一个自定义架构来将您的rdd 转换为DataFrame，并使params 列成为StructType()。您不能使用MapType()，因为所有键的值必须是相同的类型。

例如：

myRdd = [[1, 'a', 'a':[1, 2]],
         [2, 'b', 'c': 1, 'd':3],
         [3, 'c', ]]

schema = StructType(
    [
        StructField('sl', IntegerType()),
        StructField('name', StringType()),
        StructField(
            'params',
            StructType(
                [
                    StructField('a', ArrayType(IntegerType())),
                    StructField('c', IntegerType()),
                    StructField('d', IntegerType())
                ]
            )
        )
    ]
)

rdd = sc.parallelize(myRdd)
df = rdd.toDF(schema)
df.show()
#+---+----+------------------------------+
#|sl |name|params                        |
#+---+----+------------------------------+
#|1  |a   |[WrappedArray(1, 2),null,null]|
#|2  |b   |[null,1,3]                    |
#|3  |c   |[null,null,null]              |
#+---+----+------------------------------+

然后要访问特定元素，您可以使用getItem() 方法或在选择中使用"."。例如，要为每一行提取键 "a" 的值，您可以执行以下任一操作：

import pyspark.sql.functions as f
df.select("sl", "name", f.col("params").getItem("a")).show()
#+---+----+--------+
#| sl|name|params.a|
#+---+----+--------+
#|  1|   a|  [1, 2]|
#|  2|   b|    null|
#|  3|   c|    null|
#+---+----+--------+

df.select("sl", "name", "params.a").show()
#+---+----+------+
#| sl|name|     a|
#+---+----+------+
#|  1|   a|[1, 2]|
#|  2|   b|  null|
#|  3|   c|  null|
#+---+----+------+

或者您可以使用".*" 将所有元素作为单独的列访问：

df.select("sl", "name", "params.*").show()
#+---+----+------+----+----+
#| sl|name|     a|   c|   d|
#+---+----+------+----+----+
#|  1|   a|[1, 2]|null|null|
#|  2|   b|  null|   1|   3|
#|  3|   c|  null|null|null|
#+---+----+------+----+----+

DataFrame 的架构是：

df.printSchema()
#root
# |-- sl: integer (nullable = true)
# |-- name: string (nullable = true)
# |-- params: struct (nullable = true)
# |    |-- a: array (nullable = true)
# |    |    |-- element: integer (containsNull = true)
# |    |-- c: integer (nullable = true)
# |    |-- d: integer (nullable = true)

正如这意味着，结构的特定成员的数据类型必须始终相同——这意味着，例如，params.a 必须是每一行的整数列表。