从列表的 RDD 创建 Spark DataFrame

Question

我有一个 rdd（我们可以称它为 myrdd），其中 rdd 中的每条记录都是以下形式：

[('column 1',value), ('column 2',value), ('column 3',value), ... , ('column 100',value)]

我想将其转换为 pyspark 中的 DataFrame - 最简单的方法是什么？

Answer 1

查看 DataFrame documentation 以使该示例适合您，但这应该有效。我假设你的 RDD 叫做 my_rdd

from pyspark.sql import SQLContext, Row
sqlContext = SQLContext(sc)

# You have a ton of columns and each one should be an argument to Row
# Use a dictionary comprehension to make this easier
def record_to_row(record):
    schema = {'column{i:d}'.format(i = col_idx):record[col_idx] for col_idx in range(1,100+1)}
    return Row(**schema)


row_rdd = my_rdd.map(lambda x: record_to_row(x))

# Now infer the schema and you have a DataFrame
schema_my_rdd = sqlContext.inferSchema(row_rdd)

# Now you have a DataFrame you can register as a table
schema_my_rdd.registerTempTable("my_table")

我在 Spark 中使用 DataFrames 的工作不多，但这应该可以解决问题

Answer 2

使用toDF方法怎么样？您只需要添加字段名称。

df = rdd.toDF(['column', 'value'])

Answer 3

@dapangmao 的回答让我找到了这个解决方案：

my_df = my_rdd.map(lambda l: Row(**dict(l))).toDF()

Answer 4

在 pyspark 中，假设您有一个名为 userDF.

的数据框

>>> type(userDF)
<class 'pyspark.sql.dataframe.DataFrame'>

让我们将其转换为 RDD（

userRDD = userDF.rdd
>>> type(userRDD)
<class 'pyspark.rdd.RDD'>

现在您可以进行一些操作并调用例如地图函数：

newRDD = userRDD.map(lambda x:{"food":x['favorite_food'], "name":x['name']})

最后，让我们从弹性分布式数据集 (RDD) 创建一个 DataFrame。

newDF = sqlContext.createDataFrame(newRDD, ["food", "name"])

>>> type(ffDF)
<class 'pyspark.sql.dataframe.DataFrame'>

就这些了。

我之前尝试调用时遇到了这条警告消息：

newDF = sc.parallelize(newRDD, ["food","name"] : 

.../spark-2.0.0-bin-hadoop2.7/python/pyspark/sql/session.py:336: UserWarning: Using RDD of dict to inferSchema is deprecated. Use pyspark.sql.Row inst  warnings.warn("Using RDD of dict to inferSchema is deprecated. "

所以不需要再这样做了...

从列表的 RDD 创建 Spark DataFrame

Creating a Spark DataFrame from an RDD of lists

dataframe

apache-spark

pyspark