PySpark RDD 线性列表如何转换为 DataFrame？

Question

我想将线性列表转换为数据框。 即给定以下列表，

a = ["a1", "a2", "a3", b1", "b2", "b3", "c1", "c2", "c3"]

预期结果是，

+--------------------+
| col1 | col2 | col3 |
+--------------------+
|  a1  |  a2  |  a3  |
|  b1  |  b2  |  b3  |
|  c1  |  c2  |  c3  |
+--------------------+

我尝试了以下方法，但出现错误。

from pyspark.sql.types import *

a = ["a1", "a2", "a3", "b1", "b2", "b3", "c1", "c2", "c3"]

rdd = sc.parallelize(a)

schema = StructType([
     StructField("a", StringType(), True),
     StructField("b", StringType(), True),
     StructField("c", StringType(), True)
     ])

df = sqlContext.createDataFrame(rdd, schema)

df.show()

最后一个 show() 语句收到错误“作业因阶段失败而中止”。 请有人告诉我解决方案？ 谢谢。

Answer 1

根据您的comment，我假设您以rdd 开头，而不是列表。

我进一步假设您正在根据rdd 的索引确定顺序。如果这些假设是正确的，您可以使用zipWithIndex() 为每条记录添加一个行号。

然后将行号除以 3（使用整数除法）以每 3 个连续记录分组。接下来使用groupByKey() 将具有相同key 的记录聚合成一个元组。

最后，放下钥匙，拨打toDF()

rdd.zipWithIndex()\
    .map(lambda row: (row[1]//3, row[0]))\
    .groupByKey()\
    .map(lambda row: tuple(row[1]))\
    .toDF(["a", "b", "c"])\
    .show()
#+---+---+---+
#|  a|  b|  c|
#+---+---+---+
#| a1| a2| a3|
#| c1| c2| c3|
#| b1| b2| b3|
#+---+---+---+

Answer 2

这是一种有望满足您的标准的方法

# First get a 1 column DF
df = sql.createDataFrame(sc.parallelize(a).map(lambda x: [x]), schema=['col'])
# split each value into a number and letter e.g. 'a1' --> ['a','1']) 
df = df.withColumn('letter', f.split('col', '').getItem(0))
df = df.withColumn('number', f.split('col', '').getItem(1))

# Now pivot to get what you want (dropping extraneous columns and ordering 
# to get exact output

output = (df.groupBy('letter')
          .pivot('number')
          .agg(f.first('col'))
          .select([f.col(column).alias('col%s'%(column)) for column in ['1','2','3']])
          .orderBy('col1')
          .drop('letter'))