如何在 pyspark 中使用 df.write.csv 附加到 csv 文件？

Question

我正在尝试使用 df.write.csv 将数据附加到我的 csv 文件中。这是我在关注 spark 文档 http://spark.apache.org/docs/2.0.1/api/python/pyspark.sql.html#pyspark.sql.DataFrameWriter 后所做的：

from pyspark.sql import DataFrameWriter
.....
df1 = sqlContext.createDataFrame(query1)
df1.write.csv("/opt/Output/sqlcsvA.csv", append) #also tried 'mode=append'

执行上面的代码给我错误：

NameError: name 'append' not defined

没有附加，错误：

路径已经存在。

Answer 1

df.write.save(path='csv', format='csv', mode='append', sep='\t')

Answer 2

来自文档： https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrameWriter 从 v1.4 开始

csv(path, mode=None, compression=None, sep=None, quote=None, escape=None, header=None, nullValue=None, escapeQuotes=None, quoteAll=None, dateFormat=None, timestampFormat=None)

例如

from pyspark.sql import DataFrameWriter
.....
df1 = sqlContext.createDataFrame(query1)
df1.write.csv(path="/opt/Output/sqlcsvA.csv", mode="append")

如果您想编写单个文件，您可以在其中任何一行上使用 coalesce 或 repartition。哪一行都没有关系，因为数据帧只是一个 DAG 执行，在写入 csv 之前不会执行任何操作。 repartition & coalesce 有效地使用相同的代码，但合并只能减少分区数量，repartition 也可以增加它们。为简单起见，我会坚持使用repartition。

例如

df1 = sqlContext.createDataFrame(query1).repartition(1)

或

df1.repartition(1).write.csv(path="/opt/Output/sqlcsvA.csv", mode="append")

我认为文档中的示例不是很好，它们没有显示使用路径以外的参数的示例。

参考你尝试过的两件事：

(append)

为此，需要有一个名为 append 的字符串变量，其中包含值“append”。 DataFrameWriter 库中没有名为 append 的字符串常量。 即你可以在你的代码中更早地添加它，然后它就可以工作了。 append = "追加"

('mode=append')

为此，csv 方法必须解析出mode=append 字符串以获取模式的值，当您可以只拥有一个值恰好为“append”或“的参数时，这将是额外的工作。覆盖”需要提取。 None 是特例，Python 内置，不是 pyspark 特有的。

另一方面，我建议尽可能使用命名参数。 例如

csv(path="/path/to/file.csv", mode="append")

而不是位置参数

csv("/path/to/file.csv", "append")

它更清晰，有助于理解。

Answer 3

我不了解 Python，但在 Scala 和 Java 中可以通过以下方式设置保存模式：

df.write.mode("append").csv("pathToFile")

我认为它在 Python 中应该是类似的。 This 可能会有所帮助。