从 python 字典创建 pyspark 数据框

Question

我想从 python 字典创建一个 pyspark 数据框，但是下面的代码

from pyspark.sql import SparkSession, Row

df_stable = spark.createDataFrame(dict_stable_feature)
df_stable.show()

显示此错误

TypeError: Can not infer schema for type: <class 'str'>

在 stackoverflow 上阅读这篇文章：

Pyspark: Unable to turn RDD into DataFrame due to data type str instead of StringType

我可以推断出问题可能是我错误地使用了 python 标准 str 而不是 StringType 并且 spark 不喜欢它。 我该怎么做才能让它发挥作用？？

编辑：

我使用此代码创建了我的字典

Create multiple lists and store them into a dictionary Python

如您所见，创建密钥是在做

cc = str(col)
vv = "_" + str(value)
cv = cc + vv

dict_stable_feature[cv] = t

而t 只是1 和0 的二进制列表。

Answer 1

让我们开始将您的 python 字典转换为正确位置的列表值列表（即用于初始化 spark 数据帧的预期数据结构之一）。

假设字典中的所有列表值的长度相同，您可以尝试以下操作。

column_names = []
dataset = None
for column_name in dict_stable_feature:
    column_names.append(column_name)
    column_values = dict_stable_feature[column_name]
    # initialize dataset ranges
    if dataset is None:
        dataset=[]
        
        for i in range(0,len(column_values)):
            dataset.append([column_values[i]])
    else:
        for ind,val in enumerate(column_values):
            dataset[ind].append(val)

my_df = sparkSession.createDataFrame(dataset,schema=column_names)

如果所有列表值的长度不同，那么您可以尝试以下方法：

max_list_length = max([len(dict_stable_feature[k]) for k in dict_stable_feature])
column_names = []
dataset = [[] for i in range(0,max_list_length)]
default_data_value = None # feel free to change
for column_name in dict_stable_feature:
    column_names.append(column_name)
    column_values = dict_stable_feature[column_name]


    for ind,val in enumerate(column_values):
        dataset[ind].append(val)

    # ensure all columns have the same amount of rows
    no_of_values = len(column_values)
    if  no_of_values < max_list_length:
        for i in range(no_of_values,max_list_length):
            dataset[i].append(default_data_value)

my_df = sparkSession.createDataFrame(dataset,schema=column_names)

让我知道这是否适合你。