将大型 Spark Dataframe 写入 csv 文件

Question

我正在使用 Spark 2.3，我需要将 Spark Dataframe 保存到 csv 文件中，我正在寻找一种更好的方法来做到这一点。查看相关/类似问题，我发现 this one，但我需要更具体的：

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

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

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

使用df.coalesce(1).write.option("header", "true").csv('mycsv.csv')会导致在每个文件中写入标题，当文件合并时，中间会有标题。我错了吗？

使用spark write 再使用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

如果 DataFrame 太大，如何避免使用 Pandas？

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

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

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

使用 df.coalesce(1).write.option("header", "true").csv('mycsv.csv') 将导致标题被写入每个文件中，并且当文件 合并后，中间会有标题。我错了吗？

在这种情况下，您将只有 1 个文件（因为您使用 coalesce(1)）。所以你不需要关心标题。相反 - 你应该关心执行器上的内存 - 你可能会在执行器上得到 OOM，因为所有数据都将被移动到该执行器。

使用 spark write 然后 hadoop getmerge 比使用更好 从性能的角度合并？

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

Answer 2

我们使用了 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>