PySpark：将临时视图转换为表格并在本地驱动器中另存为 .CSV

Posted 2023-04-15

【中文标题】PySpark：将临时视图转换为表格并在本地驱动器中另存为 .CSV

【英文标题】：PySpark: Converting a Temporary View to a Table and save as .CSV in local drive

【发布时间】：2019-04-01 05:47:10

【问题描述】：

我正在 PySpark 2.4 中创建 JDBC 查询的临时视图。我的数据源是MS SQL Server 2017。

df = spark.read.format("jdbc").options(url="url",properties =  "driver": "com.microsoft.sqlserver.jdbc.SQLServerDriver" ,dbtable="dbtable").load()
df.createOrReplaceTempView("df_temp")

现在我可以查询临时创建的表

df_new = spark.sql("select * from df_temp where ...#standard query")

现在我想将上面的df_new 写为CSV 在我的本地驱动器中。一种方法（目前我正在做同样的事情）是转换 df_new.toPandas()。然后将其保存为csv（标准df.to_csv('/path/file.csv' 方法）。此方法创建一个名为file.csv 的文件夹，并在此文件夹中生成一个名称以part-00000-fd4c62bd-f208-4bd3-ae99-f81338b9ede1-c000.csv 开头的csv 文件。

所以如果我每天运行我的.py 文件（使用像crontab 这样的调度程序），这绝对不是一个好的选择。

问题：

如何标准化 .csv 文件的名称，并且可以每天将新数据附加到同一文件中？

有什么简单的方法可以将df_new 转换为表格或Spark DataFrame。这样我就可以申请df.coalesce(1).option("header","true").csv('/path/myfile.csv')？

【问题讨论】：

df_new 是一个数据框，因此您可以执行coalesce(1) 并保存到本地路径。

我收到错误，因为“NoneObject 没有属性写入”。

@pythondumb，您只能指定要将文件写入的文件夹名称。您无法控制正在创建的 json 文件的名称。参考这个链接，***.com/questions/41990086/…

@Gladiator：假设我使用 python 等效技术（如此处所述，***.com/questions/48652050/… 这种情况下如何每天将新数据附加到同一个文件？

我不确定你为什么要拥有单个文件，因为它会影响并行性而不是最佳实践。但是，有很多技巧。您可以尝试，1. 将所有当前数据移动到临时位置 2. 读取此临时位置并合并（1）并写入目标

【参考方案1】：

使用它来保存到本地文件系统

    #for Output in multiple files:
    df.write.option("header", "true").csv("/path/output.csv")

    #for output in single file:
    df.coalesce(1).write.option("header", "true").csv("/path/output.csv")