如何在不使用循环的情况下从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 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|
+---+---+------+

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

Answer 1

当您意识到您的初始列表为 nested 时。即，实际列表是外部元素的独特元素，然后您会看到该解决方案仅通过考虑其第一个（又是）元素而容易出现：

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 flatten and reshape

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|
+---+-----+---+