pyspark - 从 RDD 管道到 DF Pyspark

Question

当我开始寻求帮助时，这个问题似乎很受欢迎，但我还没有找到解决方案。事实上，你可能会发现一些你认为可能重复的，但我想我在过去的几个小时里都试过了。据我了解，sqlContext 将在这里解决问题，但我对任何有效的答案持开放态度。我正在使用 Spark 2.1

我从我从 mongodb 中提取的 id 列表开始。
示例输出:

[u'182028', u'161936', u'12333', u'120677']
'rated_game_ids_lst type:' <type 'list'>

然后我继续尝试创建一个我想变成 DF 的 RDD:

user_unrated_games = ugr_rdd.filter(lambda x: x[1] not in rated_game_ids_lst).map(lambda x: (19, x[1], x[2]))

示例输出:

'user_unrated_games:' [(19, u'174430', 3.4), (19, u'169786', 3.4)]
'user_unrated_games type:' <class 'pyspark.rdd.PipelinedRDD'>

以及我上面使用的 urg_rdd 示例(第一行):

'ugr_rdd:'[Row(user_id=5, game_id=u'182028', rating=9.15)]
'ugr_rdd_type:' pyspark.rdd.RDD

然后我试试这个:

df = sqlContext.createDataFrame(user_unrated_games, ['user_id', 'game_id', 'rating'])

该方法失败，所以我尝试了这个:

user_unrated_games = ugr_rdd.filter(lambda x: x[1] not in rated_game_ids_lst).map(lambda x: Row(user_id=19, game_id=x[1], rating= x[2]))

示例输出:

('user_unrated_games type:', <class 'pyspark.rdd.PipelinedRDD'>)
('user_unrated_games:', [Row(game_id=u'174430', rating=3.4, user_id=19), Row(game_id=u'169786', rating=3.4, user_id=19)])

然后这个:

df = sqlContext.createDataFrame(user_unrated_games)

这两种尝试都会出现此错误:

IllegalArgumentException: u"Error while instantiating 'org.apache.spark.sql.hive.HiveSessionState':"

从那里，我开始尝试更改“user_id”等类型的组合，尝试按原样传递 RDD，尝试将我的管道转换为 RDD……坦率地说，我尝试了很多东西，但上面两个看起来最接近似乎对其他人有用的东西。

Answer 1

问题是您需要在调用 createDataFrame 时指定包含数据类型的架构。方法。像这样的事情应该可以解决问题:

from pyspark.sql.types import *

rdd = sc.parallelize([(19, 174430, 3.4), (19, 169786, 3.4)])

schema = StructType( [
    StructField('user_id', IntegerType()),
    StructField('game_id', IntegerType()),
    StructField('rating', FloatType())
    ])

df = spark.createDataFrame(rdd, schema)

df.show()

注意:我已经使用 spark 2.1.0 对此进行了测试。在这种情况下 spark是 SparkSession目的。