将标准 python 键值字典列表转换为 pyspark 数据框

Posted 2023-04-15

【中文标题】将标准 python 键值字典列表转换为 pyspark 数据框

【英文标题】：Convert a standard python key value dictionary list to pyspark data frame

【发布时间】：2016-06-02 06:17:50

【问题描述】：

假设我有一个 python 字典键值对列表，其中键对应于表的列名，所以下面列出了如何将其转换为具有两个 cols arg1 arg2 的 pyspark 数据帧？

 ["arg1": "", "arg2": "","arg1": "", "arg2": "","arg1": "", "arg2": ""]

我怎样才能使用以下构造来做到这一点？

df = sc.parallelize([
    ...
]).toDF

上面代码中arg1 arg2的放置位置(...)

【问题讨论】：

你应该编辑你的问题，而不是“...”，请告诉我们“arg1”和“arg2”应该去哪里。

@betterworld ok 完成了怎么办

【参考方案1】：

老办法：

sc.parallelize(["arg1": "", "arg2": "","arg1": "", "arg2": "","arg1": "", "arg2": ""]).toDF()

新方法：

from pyspark.sql import Row
from collections import OrderedDict

def convert_to_row(d: dict) -> Row:
    return Row(**OrderedDict(sorted(d.items())))

sc.parallelize(["arg1": "", "arg2": "","arg1": "", "arg2": "","arg1": "", "arg2": ""]) \
    .map(convert_to_row) \ 
    .toDF()

【讨论】：

谢谢，能否请您回答相关问题：***.com/questions/37584185/…

这不是 scala 吗？ def convert_to_row(d: dict) -> 行：

@rado 那是 Python 3 的函数注解。

@Andre85 我认为因为每个字典中键的顺序可能不同，所以我们需要排序。

如果缺少键会发生什么，我们会得到空值还是错误。

【参考方案2】：

对于任何寻求不同解决方案的人，我发现这对我有用： 我有一个带有键值对的字典 - 我希望将其转换为两个 PySpark 数据框列：

所以

k1:v1, k2:v2 ...

变成

 ---------------- 
| col1   |  col2 |
|----------------|
| k1     |  v1   |
| k2     |  v2   |
 ----------------

lol= list(map(list, mydict.items()))
df = spark.createDataFrame(lol, ["col1", "col2"])

【讨论】：

更简单：df = spark.createDataFrame(mydict.items(), ["col1", "col2"])

【参考方案3】：

我必须修改接受的答案，以便它在运行 Spark 2.0 的 Python 2.7 中为我工作。

from collections import OrderedDict
from pyspark.sql import SparkSession, Row

spark = (SparkSession
        .builder
        .getOrCreate()
    )

schema = StructType([
    StructField('arg1', StringType(), True),
    StructField('arg2', StringType(), True)
])

dta = ["arg1": "", "arg2": "", "arg1": "", "arg2": ""]

dtaRDD = spark.sparkContext.parallelize(dta) \
    .map(lambda x: Row(**OrderedDict(sorted(x.items()))))

dtaDF = spark.createDataFrame(dtaRdd, schema) 

【参考方案4】：

其他答案有效，但这里还有一个可以很好地处理嵌套数据的单行代码。它可能不是最有效的，但是如果您从内存中的字典创建 DataFrame，那么您要么使用测试数据等小型数据集，要么使用错误的 spark，因此效率真的不应该成为问题：

d = any json compatible dict
spark.read.json(sc.parallelize([json.dumps(d)]))

【参考方案5】：

假设你的数据是一个结构体而不是一个字符串字典，你可以这样做

newdf = df.select(['df.arg1','df.arg2'])