如何在不使用 for 循环的情况下从 pyspark 中的列表创建数据框？

Question

我有如下列表：

rrr=[[(1,(3,1)),(2, (3,2)),(3, (3, 2)),(1,(4,1)),(2, (4,2))]]

df_input = []

接下来我定义了如下标题：

df_header=['sid', 'tid', 'srank']

使用 for 循环将数据附加到空列表中：

for i in rrr:
     for j in i:
            df_input.append((j[0], j[1][0], j[1][1]))
df_input

Output : [(1, 3, 1), (2, 3, 2), (3, 3, 2)]

如下创建数据框：

   df = spark.createDataFrame(df_input, df_header)
    df.show()

+---+---+------+
| sid|tid|srank|
+---+---+------+
|  1|  3|     1|
|  2|  3|     2|
|  3|  3|     2|
+---+---+------+

现在我的问题是如何在不使用任何外部 for 循环（如上）的情况下创建数据框。输入列表包含多于 1 条缺失记录。

Answer 1

当你意识到你的初始列表是一个嵌套的。即作为外部元素的唯一元素的实际列表，那么您会发现解决方案很容易通过仅考虑其第一个（也是唯一的）元素来实现：

spark.version
#  u'2.1.1'

from pyspark.sql import Row

# your exact data:
rrr=[[(1,(3,1)),(2, (3,2)),(3, (3, 2)),(1,(4,1)),(2, (4,2))]]
df_header=['sid', 'tid', 'srank']

df = sc.parallelize(rrr[0]).map(lambda x: Row(x[0], x[1][0],x[1][1])).toDF(schema=df_header)
df.show()
# +---+---+-----+ 
# |sid|tid|srank|
# +---+---+-----+
# |  1|  3|    1|
# |  2|  3|    2|
# |  3|  3|    2|
# |  1|  4|    1| 
# |  2|  4|    2|
# +---+---+-----+

Answer 2

解决方案一：引入 toDF() 转换（但修改输入）

from pyspark.sql import Row    
ar=[[1,(3,1)],[2, (3,2)],[3, (3,2)]]
sc.parallelize(ar).map(lambda x: Row(sid=x[0], tid=x[1][0],srank=x[1][1])).toDF().show()

+---+-----+---+
|sid|srank|tid|
+---+-----+---+
|  1|    1|  3|
|  2|    2|  3|
|  3|    2|  3|
+---+-----+---+

解决方案 2：使用请求的输入矩阵使用列表理解、numpy 展平和重塑

import numpy as np 
x=[[(1,(3,1)),(2, (3,2)),(3, (3, 2))]]
ar=[[(j[0],j[1][0],j[1][1]) for j in i] for i in x]
flat=np.array(ar).flatten()
flat=flat.reshape(len(flat)/3, 3)
sc.parallelize(flat).map(lambda x: Row(sid=int(x[0]),tid=int(x[1]),srank=int(x[2]))).toDF().show()

+---+-----+---+
|sid|srank|tid|
+---+-----+---+
|  1|    1|  3|
|  2|    2|  3|
|  3|    2|  3|
+---+-----+---+

#works also with N,M matrix
number_columns=3
x=[[(1,(3,1)),(2, (3,2)),(3, (3, 2))],[(5,(6,7)),(8, (9,10)),(11, (12, 13))]]
ar=[[(j[0],j[1][0],j[1][1]) for j in i] for i in x]
flat=np.array(ar).flatten()
flat=flat.reshape(int(len(flat)/number_columns), number_columns)
sc.parallelize(flat).map(lambda x: Row(sid=int(x[0]),tid=int(x[1]),srank=int(x[2]))).toDF().show()
+---+-----+---+
|sid|srank|tid|
+---+-----+---+
|  1|    1|  3|
|  2|    2|  3|
|  3|    2|  3|
|  5|    7|  6|
|  8|   10|  9|
| 11|   13| 12|
+---+-----+---+