如何在PySpark中从RDD创建数据帧？

Question

我有一个RDD看起来像这样

[((0, Row(event_type_new=u'ALERT|VEHICLE_HEALTH_DATA|CHANGE_IN_HEALTH|DTC|B109F|', day=u'Fri')), 0), 
 ((1, Row(event_type_new=u'ALERT|VEHICLE_HEALTH_DATA|CHANGE_IN_HEALTH|DTC|B1115|HIGH MOUNTED STOP LAMP CONTROL', day=u'Sat')), 2)]

它有一个索引，一个Row对象（event_type_new和day），后跟一个预测（整数）。我如何创建一个包含3列的数据帧，包括event_type_new，day，和Prediction。在

我正在使用spark1.6.2和pysparkapi。在

谢谢！在

Answer 1

我假设这是一个collected RDD，因为看起来你得到了一个list和{}和{}对象组合的元组。您可以通过以下方式获得所需的输出：

from pyspark.sql import Row


lst = [((0, Row(event_type_new=u'ALERT|VEHICLE_HEALTH_DATA|CHANGE_IN_HEALTH|DTC|B109F|', day=u'Fri')), 0),
       ((1, Row(event_type_new=u'ALERT|VEHICLE_HEALTH_DATA|CHANGE_IN_HEALTH|DTC|B1115|HIGH MOUNTED STOP LAMP CONTROL', day=u'Sat')), 2)]

output = []
for row in lst:
    vals = tuple(row[0][1]) + (row[1],)
    fields = row[0][1].__fields__ + ['prediction']
    row = Row(*vals)
    row.__fields__ = fields
    output.append(row)

df = sc.parallelize(output).toDF()
df.show()

您应该得到如下内容：

^{pr2}$

我希望这有帮助。在

Answer 2

首先将列表转换为RDD。然后将每个元素映射到Row。您可以使用.toDF()方法轻松地将Row的列表转换为数据帧

from pyspark.sql import Row

ls = [((0, Row(event_type_new=u'ALERT|VEHICLE_HEALTH_DATA|CHANGE_IN_HEALTH|DTC|B109F|', day=u'Fri')), 0),
      ((1, Row(event_type_new=u'ALERT|VEHICLE_HEALTH_DATA|CHANGE_IN_HEALTH|DTC|B1115|HIGH MOUNTED STOP LAMP CONTROL', day=u'Sat')), 2)]
ls_rdd = sc.parallelize(ls)
ls_row = ls_rdd.map(lambda x: Row(**{'day': str(x[0][1].day), 'event_type': str(x[0][1].event_type_new), 'prediction': int(x[1])}))
df = ls_row.toDF()

当您运行df.show()时，它将如下所示：

^{pr2}$