使用 spark-csv 编写单个 CSV 文件

Question

我正在使用 https://github.com/databricks/spark-csv ，我正在尝试编写单个 CSV，但不能，它正在创建一个文件夹。

需要一个 Scala 函数，该函数将接受路径和文件名等参数并写入该 CSV 文件。

Answer 1

它正在创建一个包含多个文件的文件夹，因为每个分区都是单独保存的。如果您需要单个输出文件（仍在文件夹中），您可以repartition（如果上游数据很大，则首选，但需要随机播放）：

df
   .repartition(1)
   .write.format("com.databricks.spark.csv")
   .option("header", "true")
   .save("mydata.csv")

或coalesce:

df
   .coalesce(1)
   .write.format("com.databricks.spark.csv")
   .option("header", "true")
   .save("mydata.csv")

保存前的数据框：

所有数据都将写入mydata.csv/part-00000。在您使用此选项之前确保您了解正在发生的事情以及将所有数据传输给单个工作人员的成本是多少。如果您将分布式文件系统与复制一起使用，数据将被多次传输——首先获取到单个工作人员，然后分布在存储节点上。

或者，您可以保留您的代码，然后使用cat 或HDFS getmerge 等通用工具在之后简单地合并所有部分。

Answer 2

如果您使用 HDFS 运行 Spark，我一直在通过正常写入 csv 文件并利用 HDFS 进行合并来解决问题。我直接在 Spark (1.6) 中这样做：

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs._

def merge(srcPath: String, dstPath: String): Unit =  {
   val hadoopConfig = new Configuration()
   val hdfs = FileSystem.get(hadoopConfig)
   FileUtil.copyMerge(hdfs, new Path(srcPath), hdfs, new Path(dstPath), true, hadoopConfig, null) 
   // the "true" setting deletes the source files once they are merged into the new output
}


val newData = << create your dataframe >>


val outputfile = "/user/feeds/project/outputs/subject"  
var filename = "myinsights"
var outputFileName = outputfile + "/temp_" + filename 
var mergedFileName = outputfile + "/merged_" + filename
var mergeFindGlob  = outputFileName

    newData.write
        .format("com.databricks.spark.csv")
        .option("header", "false")
        .mode("overwrite")
        .save(outputFileName)
    merge(mergeFindGlob, mergedFileName )
    newData.unpersist()

不记得我在哪里学的这个技巧，但它可能对你有用。

Answer 3

我在这里玩游戏可能有点晚了，但是使用coalesce(1) 或repartition(1) 可能适用于小型数据集，但大型数据集将全部放入一个节点上的一个分区中。这可能会引发 OOM 错误，或者充其量是处理缓慢。

我强烈建议您使用 Hadoop API 中的 FileUtil.copyMerge() 函数。这会将输出合并到一个文件中。

EDIT - 这有效地将数据带到驱动程序而不是执行程序节点。 Coalesce() 如果单个 executor 的 RAM 比驱动程序多，那就没问题了。

编辑 2：copyMerge() 在 Hadoop 3.0 中被删除。有关如何使用最新版本的更多信息，请参阅以下堆栈溢出文章：How to do CopyMerge in Hadoop 3.0?

Answer 4

如果您正在使用 Databricks 并且可以将所有数据放入一个工作人员的 RAM（因此可以使用 .coalesce(1)），您可以使用 dbfs 来查找并移动生成的 CSV 文件：

val fileprefix= "/mnt/aws/path/file-prefix"

dataset
  .coalesce(1)       
  .write             
//.mode("overwrite") // I usually don't use this, but you may want to.
  .option("header", "true")
  .option("delimiter","\t")
  .csv(fileprefix+".tmp")

val partition_path = dbutils.fs.ls(fileprefix+".tmp/")
     .filter(file=>file.name.endsWith(".csv"))(0).path

dbutils.fs.cp(partition_path,fileprefix+".tab")

dbutils.fs.rm(fileprefix+".tmp",recurse=true)

如果您的文件不适合工作人员的 RAM，您可能需要考虑 chaotic3quilibrium's suggestion to use FileUtils.copyMerge()。我还没有这样做，也不知道是否可行，例如在 S3 上。

这个答案是建立在以前对这个问题的答案以及我自己对提供的代码 sn-p 的测试之上的。 I originally posted it to Databricks 并在这里重新发布。

我找到的关于 dbfs 的 rm 递归选项的最佳文档位于 a Databricks forum。

Answer 5

spark 的 df.write() API 将在给定路径内创建多个部分文件...强制 spark 只写入单个部分文件使用 df.coalesce(1).write.csv(...) 而不是 df.repartition(1).write.csv(...) 因为合并是一个狭窄的转换，而重新分区是一个广泛的转换参见Spark - repartition() vs coalesce()

df.coalesce(1).write.csv(filepath,header=True) 

将在给定的文件路径中创建包含一个part-0001-...-c000.csv 文件的文件夹 使用

cat filepath/part-0001-...-c000.csv > filename_you_want.csv 

要有一个用户友好的文件名

Answer 6

我在 Python 中使用它来获取单个文件：

df.toPandas().to_csv("/tmp/my.csv", sep=',', header=True, index=False)

Answer 7

此答案扩展了已接受的答案，提供了更多上下文，并提供了可以在您机器上的 Spark Shell 中运行的代码 sn-ps。

有关已接受答案的更多上下文

接受的答案可能会给您这样的印象，示例代码输出单个 mydata.csv 文件，但事实并非如此。让我们演示一下：

val df = Seq("one", "two", "three").toDF("num")
df
  .repartition(1)
  .write.csv(sys.env("HOME")+ "/Documents/tmp/mydata.csv")

这是输出的内容：

Documents/
  tmp/
    mydata.csv/
      _SUCCESS
      part-00000-b3700504-e58b-4552-880b-e7b52c60157e-c000.csv

注意mydata.csv 是已接受答案中的文件夹 - 它不是文件！

如何输出具有特定名称的单个文件

我们可以使用spark-daria 写出一个mydata.csv 文件。

import com.github.mrpowers.spark.daria.sql.DariaWriters
DariaWriters.writeSingleFile(
    df = df,
    format = "csv",
    sc = spark.sparkContext,
    tmpFolder = sys.env("HOME") + "/Documents/better/staging",
    filename = sys.env("HOME") + "/Documents/better/mydata.csv"
)

这将输出如下文件：

Documents/
  better/
    mydata.csv

S3 路径

您需要将 s3a 路径传递给 DariaWriters.writeSingleFile 才能在 S3 中使用此方法：

DariaWriters.writeSingleFile(
    df = df,
    format = "csv",
    sc = spark.sparkContext,
    tmpFolder = "s3a://bucket/data/src",
    filename = "s3a://bucket/data/dest/my_cool_file.csv"
)

请参阅here 了解更多信息。

避免复制合并

copyMerge 已从 Hadoop 3 中删除。DariaWriters.writeSingleFile 实现使用 fs.rename、as described here。 Spark 3 still used Hadoop 2，所以 copyMerge 实现将在 2020 年工作。我不确定 Spark 何时会升级到 Hadoop 3，但最好避免在 Spark 升级 Hadoop 时导致代码中断的任何 copyMerge 方法。

源代码

如果您想检查实现，请在 spark-daria 源代码中查找 DariaWriters 对象。

PySpark 实施

使用 PySpark 写出单个文件更容易，因为您可以将 DataFrame 转换为默认情况下作为单个文件写出的 Pandas DataFrame。

from pathlib import Path
home = str(Path.home())
data = [
    ("jellyfish", "JALYF"),
    ("li", "L"),
    ("luisa", "LAS"),
    (None, None)
]
df = spark.createDataFrame(data, ["word", "expected"])
df.toPandas().to_csv(home + "/Documents/tmp/mydata-from-pyspark.csv", sep=',', header=True, index=False)

限制

DariaWriters.writeSingleFile Scala 方法和df.toPandas() Python 方法仅适用于小型数据集。巨大的数据集不能写成单个文件。从性能的角度来看，将数据作为单个文件写入并不是最佳的，因为数据不能并行写入。

Answer 8

适用于从 Minkymorgan 修改的 S3 的解决方案。

只需将临时分区目录路径（名称与最终路径不同）传递为 srcPath，将单个最终 csv/txt 传递为 destPath，如果要删除原始目录，还要指定 deleteSource。

/**
* Merges multiple partitions of spark text file output into single file. 
* @param srcPath source directory of partitioned files
* @param dstPath output path of individual path
* @param deleteSource whether or not to delete source directory after merging
* @param spark sparkSession
*/
def mergeTextFiles(srcPath: String, dstPath: String, deleteSource: Boolean): Unit =  {
  import org.apache.hadoop.fs.FileUtil
  import java.net.URI
  val config = spark.sparkContext.hadoopConfiguration
  val fs: FileSystem = FileSystem.get(new URI(srcPath), config)
  FileUtil.copyMerge(
    fs, new Path(srcPath), fs, new Path(dstPath), deleteSource, config, null
  )
}

Answer 9

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs._
import org.apache.spark.sql.{DataFrame,SaveMode,SparkSession}
import org.apache.spark.sql.functions._

我使用以下方法解决了（hdfs 重命名文件名）：-

第 1 步：-（创建数据帧并写入 HDFS）

df.coalesce(1).write.format("csv").option("header", "false").mode(SaveMode.Overwrite).save("/hdfsfolder/blah/")

第 2 步：-（创建 Hadoop 配置）

val hadoopConfig = new Configuration()
val hdfs = FileSystem.get(hadoopConfig)

Step3 :- (获取hdfs文件夹路径)

val pathFiles = new Path("/hdfsfolder/blah/")

Step4:- (从 hdfs 文件夹中获取 spark 文件名)

val fileNames = hdfs.listFiles(pathFiles, false)
println(fileNames)

setp5:- (创建 scala 可变列表以保存所有文件名并将其添加到列表中)

    var fileNamesList = scala.collection.mutable.MutableList[String]()
    while (fileNames.hasNext) {
      fileNamesList += fileNames.next().getPath.getName
    }
    println(fileNamesList)

第6步：-（从文件名scala列表中过滤_SUCESS文件顺序）

    // get files name which are not _SUCCESS
    val partFileName = fileNamesList.filterNot(filenames => filenames == "_SUCCESS")

步骤 7：-（将 scala 列表转换为字符串并将所需文件名添加到 hdfs 文件夹字符串，然后应用重命名）

val partFileSourcePath = new Path("/yourhdfsfolder/"+ partFileName.mkString(""))
    val desiredCsvTargetPath = new Path(/yourhdfsfolder/+ "op_"+ ".csv")
    hdfs.rename(partFileSourcePath , desiredCsvTargetPath)

Answer 10

在保存之前重新分区/合并到 1 个分区（您仍然会得到一个文件夹，但其中会有一个部分文件）

Answer 11

你可以使用rdd.coalesce(1, true).saveAsTextFile(path)

它将数据作为单个文件存储在 path/part-00000 中

Answer 12

spark.sql("select * from df").coalesce(1).write.option("mode","append").option("header","true").csv("/your/hdfs/path/")

spark.sql("select * from df") --> 这是数据框

coalesce(1) 或 repartition(1) --> 这将使您的输出文件仅成为 1 个部分文件

write --> 写入数据

option("mode","append") --> 将数据附加到现有目录

option("header","true") --> 启用标头

csv("<hdfs dir>") --> 写入为 CSV 文件及其在 HDFS 中的输出位置

Answer 13

通过使用 Listbuffer，我们可以将数据保存到单个文件中：

import java.io.FileWriter
import org.apache.spark.sql.SparkSession
import scala.collection.mutable.ListBuffer
    val text = spark.read.textFile("filepath")
    var data = ListBuffer[String]()
    for(line:String <- text.collect()){
      data += line
    }
    val writer = new FileWriter("filepath")
    data.foreach(line => writer.write(line.toString+"\n"))
    writer.close()

Answer 14

def export_csv(  
  fileName: String,
  filePath: String
  ) = {

  val filePathDestTemp = filePath + ".dir/"
  val merstageout_df = spark.sql(merstageout)

  merstageout_df
    .coalesce(1)
    .write
    .option("header", "true")
    .mode("overwrite")
    .csv(filePathDestTemp)
  
  val listFiles = dbutils.fs.ls(filePathDestTemp)

  for(subFiles <- listFiles){
      val subFiles_name: String = subFiles.name
      if (subFiles_name.slice(subFiles_name.length() - 4,subFiles_name.length()) == ".csv") {
        dbutils.fs.cp (filePathDestTemp + subFiles_name,  filePath + fileName+ ".csv")
        dbutils.fs.rm(filePathDestTemp, recurse=true)
      }}} 

Answer 15

还有另一种使用 Java 的方法

import java.io._

def printToFile(f: java.io.File)(op: java.io.PrintWriter => Unit) 
  {
     val p = new java.io.PrintWriter(f);  
     try { op(p) } 
     finally { p.close() }
  } 

printToFile(new File("C:/TEMP/df.csv")) { p => df.collect().foreach(p.println)}