将列表转换为 pyspark 数据框

Posted 2023-04-15

【中文标题】将列表转换为 pyspark 数据框

【英文标题】：Convert list into pyspark dataframe

【发布时间】：2018-06-28 06:18:00

【问题描述】：

我有一个从 mysql 数据库中获取的元组列表。我需要将其转换为 pyspark 数据框。

我的代码如下所示：

os.environ['PYSPARK_SUBMIT_ARGS'] = "--packages mysql:mysql-connector-java:5.1.39 pyspark-shell"

    spark = SparkSession.builder.appName('recommendation_clustering').getOrCreate()
    sqlContext = SQLContext(spark)
    final_result = sqlContext.read.format("jdbc").options(
        url="jdbc:mysql://localhost:3306/mysqldb",
        driver="com.mysql.jdbc.Driver",
        dbtable="(select movie_id, genres from  program) as rating",
        user="user",
        password="password",
        properties="driver": 'com.mysql.jdbc.Driver'
    ).load().take(3)

final_result的类型是list，如下所示：

[行（movie_id='0-0-10'，流派='[喜剧]'），行（movie_id='0-0-1113'， 流派='[音乐]'), 行(movie_id='0-0-1132', 流派='[音乐]')])

我需要将其转换为 pyspark 数据框。

我试过sqlDataFrame = sqlContext.createDataFrame(final_result,["movie_id","genres"]) 但收到错误消息。

【问题讨论】：

错误是什么？

AttributeError: 'SparkSession' 对象没有属性 'parallelize'

您需要使用take(3) 吗？因为没有它，final_result 已经是一个 pyspark 数据帧。

我以为有一个数据框，而不是 pyspark 数据框

【参考方案1】：

您在 cmets 中提到的错误 AttributeError: 'SparkSession' object has no attribute 'parallelize' 可能是因为您尝试使用 SparkSession 创建一个 SQLContext。 SparkSession 是使用 Spark 操作数据的统一入口点，您无需单独创建 SQLContext。直接使用spark变量读取数据。

阅读this answer了解更多详情。

此外，如果您只使用 spark 而不是 sqlContext，那么您已经在 pyspark 数据框中获取数据 final_result。

【讨论】：

当我删除 take 函数时，我得到的输出为 DataFrame[movie_id: string,genres: string]

是的！那是你的 pyspark 数据框。写final_result.show()查看数据。

知道了。我是新来的火花。非常感谢。