将big Spark数据帧写入csv文件

Question

我正在使用Spark 2.3，我需要将Spark数据帧保存到csv文件中，我正在寻找更好的方法。。查看相关/类似的问题，我发现了这个问题，但我需要一个更具体的：

如果DataFrame太大，如何避免使用Pandas？因为我使用了toCSV（）函数（下面的代码），它产生了：

内存不足错误（无法分配内存）。

使用文件I/O直接写入csv是更好的方法吗？它可以保留分隔符吗？

使用df。聚结（1）。写选项（“标题”、“true”）。csv（'mycsv.csv'）将导致在每个文件中写入标题，当文件合并时，标题将位于中间在中间。我错了吗？

从性能角度来看，使用spark编写hadoop和getmerge比使用coalesce更好？

def toCSV(spark_df, n=None, save_csv=None, csv_sep=',', csv_quote='"'):
        """get spark_df from hadoop and save to a csv file

        Parameters
        ----------
        spark_df: incoming dataframe
        n: number of rows to get
        save_csv=None: filename for exported csv

        Returns
        -------

        """

        # use the more robust method
        # set temp names
        tmpfilename = save_csv or (wfu.random_filename() + '.csv')
        tmpfoldername = wfu.random_filename()
        print n
        # write sparkdf to hadoop, get n rows if specified
        if n:
            spark_df.limit(n).write.csv(tmpfoldername, sep=csv_sep, quote=csv_quote)
        else:
            spark_df.write.csv(tmpfoldername, sep=csv_sep, quote=csv_quote)

        # get merge file from hadoop
        HDFSUtil.getmerge(tmpfoldername, tmpfilename)
        HDFSUtil.rmdir(tmpfoldername)

        # read into pandas df, remove tmp csv file
        pd_df = pd.read_csv(tmpfilename, names=spark_df.columns, sep=csv_sep, quotechar=csv_quote)
        os.remove(tmpfilename)

        # re-write the csv file with header!
        if save_csv is not None:
            pd_df.to_csv(save_csv, sep=csv_sep, quotechar=csv_quote)

Answer 1

我们使用了databricks库。很好用

df.save("com.databricks.spark.csv", SaveMode.Overwrite, Map("delimiter" -> delim, "nullValue" -> "-", "path" -> tempFPath))

图书馆：

<!-- spark df to csv -->
    <dependency>
        <groupId>com.databricks</groupId>
        <artifactId>spark-csv_2.10</artifactId>
        <version>1.3.0</version>
    </dependency>

Answer 2

如果数据框太大，如何避免使用熊猫？

您只需将文件保存到HDFS或S3或您拥有的任何分布式存储中即可。

使用文件I/O直接写入csv是更好的方法吗？它能保存分离器吗？

如果您的意思是将文件保存到本地存储，它仍然会导致OOM异常，因为您需要移动本地计算机上内存中的所有数据才能执行此操作。

使用df。聚结（1）。写选项（“标题”、“true”）。csv（'mycsv.csv'）将导致在每个文件中写入标题，并且当文件合并时，标题将在中间。我错了吗？

在这种情况下，您将只有1个文件（因为您执行coalesce（1））。所以您不需要关心标头。相反——您应该关心执行程序上的内存——您可能会在执行程序上获得OOM，因为所有数据都将移动到该执行程序。

从性能的角度来看，使用火花写入然后使用hadoop getmer比使用合并更好？

当然更好（但不要使用coalesce（））。Spark将高效地将数据写入存储器，然后HDFS将复制数据，然后getmerge将能够高效地从节点读取数据并将其合并。