Spark 应用程序中如何从 DataFrame (Scala) 创建 CSV 文件？

Question

我的下一个问题并不新鲜，但我想了解如何一步一步完成。

在 Spark 应用程序中，我创建了 DataFrame。让我们称之为df。 Spark 版本：2.4.0

val df: DataFrame  = Seq(
    ("Alex", "2018-01-01 00:00:00", "2018-02-01 00:00:00", "OUT"),
    ("Bob", "2018-02-01 00:00:00", "2018-02-05 00:00:00", "IN"),
    ("Mark", "2018-02-01 00:00:00", "2018-03-01 00:00:00", "IN"),
    ("Mark", "2018-05-01 00:00:00", "2018-08-01 00:00:00", "OUT"),
    ("Meggy", "2018-02-01 00:00:00", "2018-02-01 00:00:00", "OUT")
).toDF("NAME", "START_DATE", "END_DATE", "STATUS")

如何从这个DataFrame创建.csv文件并将csv文件放到服务器的特定文件夹中？

例如，这段代码是否正确？我注意到有些人使用coalesce 或repartition 来完成这项任务。但我不明白在我的情况下哪一个会更好。

union.write
  .format("com.databricks.spark.csv")
  .option("header", "true")
  .save("/home/reports/")

当我尝试使用下一个代码时，它会引发 ERROR:

org.apache.hadoop.security.AccessControlException: Permission denied: user=root, access=WRITE, inode="/home/reports/_temporary/0":hdfs:hdfs:drwxr-xr-x 

我以root 用户身份运行 Spark 应用程序。 reports 由root 用户使用下一条命令创建的文件夹：

mkdir -m 777 reports

好像只有hdfs用户可以写文件。

Answer 1

我相信您对 Spark 的行为方式感到困惑，我建议您先阅读官方文档和/或一些教程。
不过，我希望这能回答您的问题。

此代码会将 DataFrame 保存为本地文件系统上的 SINGLE CSV 文件。
已使用 Spark 2.4.0 与 Scala 2.12.8 在 Ubuntu 18.04 笔记本电脑上。

import org.apache.spark.sql.SparkSession

val spark =
  SparkSession
    .builder
    .master("local[*]")
    .appName("CSV Writter Test")
    .getOrCreate()
import spark.implicits._

val df =
  Seq(
    ("Alex", "2018-01-01 00:00:00", "2018-02-01 00:00:00", "OUT"),
    ("Bob", "2018-02-01 00:00:00", "2018-02-05 00:00:00", "IN"),
    ("Mark", "2018-02-01 00:00:00", "2018-03-01 00:00:00", "IN"),
    ("Mark", "2018-05-01 00:00:00", "2018-08-01 00:00:00", "OUT"),
    ("Meggy", "2018-02-01 00:00:00", "2018-02-01 00:00:00", "OUT")
  ).toDF("NAME", "START_DATE", "END_DATE", "STATUS")

df.printSchema
// root
//  |-- NAME: string (nullable = true)
//  |-- START_DATE: string (nullable = true)
//  |-- END_DATE: string (nullable = true)
//  |-- STATUS: string (nullable = true)

df.coalesce(numPartitions = 1)
  .write
  .option(key = "header", value = "true")
  .option(key = "sep", value = ",")
  .option(key = "encoding", value = "UTF-8")
  .option(key = "compresion", value = "none")
  .mode(saveMode = "OVERWRITE")
  .csv(path = "file:///home/balmungsan/dailyReport/") // Change the path. Note there are 3 /, the first two are for the file protocol, the third one is for the root folder.

spark.stop()

现在，让我们检查保存的文件。

balmungsan@BalmungSan:dailyReport $ pwd
/home/balmungsan/dailyReport

balmungsan@BalmungSan:dailyReport $ ls
part-00000-53a11fca-7112-497c-bee4-984d4ea8bbdd-c000.csv  _SUCCESS

balmungsan@BalmungSan:dailyReport $ cat part-00000-53a11fca-7112-497c-bee4-984d4ea8bbdd-c000.csv 
NAME,START_DATE,END_DATE,STATUS
Alex,2018-01-01 00:00:00,2018-02-01 00:00:00,OUT
Bob,2018-02-01 00:00:00,2018-02-05 00:00:00,IN
Mark,2018-02-01 00:00:00,2018-03-01 00:00:00,IN
Mark,2018-05-01 00:00:00,2018-08-01 00:00:00,OUT
Meggy,2018-02-01 00:00:00,2018-02-01 00:00:00,OUT

_SUCCESS 文件的存在表示写入成功。

重要提示：

• 您需要指定file:// 协议以保存到本地文件系统，而不是HDFS。
• 路径指定保存文件分区的文件夹的名称，而不是文件的名称，在该文件夹中，每个分区将有一个文件。如果你想用Spark再次读取这样的文件，那么你只需要指定文件夹，Spark就会理解分区文件。如果没有，我建议在之后重命名文件 - 据我所知，没有办法从 Spark 控制名称。
• 如果 df 太大而无法仅容纳一个节点的内存，则作业将失败。
• 如果您以分布式方式运行它（例如使用主纱线），则文件将不会保存在主节点中，而是保存在其中一个从节点中。如果你真的需要它在主节点中，那么你可以收集它并按照 Dmitry 的建议用普通的 Scala 编写它。