python - 使用(spark、python、pyspark、jupyter)将多个项目保存到 HDFS

Question

我习惯用 Python 编程。我的公司现在安装了一个安装了 Jupyter 的 Hadoop 集群。到目前为止，我从未使用过 Spark/Pyspark 做任何事情。

我可以像这样简单地从 HDFS 加载文件:

text_file = sc.textFile("/user/myname/student_grades.txt")

我可以像这样编写输出:

text_file.saveAsTextFile("/user/myname/student_grades2.txt")

我想要实现的目标是使用一个简单的“for 循环”来逐个读取文本文件并将其内容写入一个 HDFS 文件中。所以我尝试了这个:

list = ['text1.txt', 'text2.txt', 'text3.txt', 'text4.txt']

for i in list:
    text_file = sc.textFile("/user/myname/" + i)
    text_file.saveAsTextFile("/user/myname/all.txt")

所以这适用于列表的第一个元素，但随后给我这个错误消息:

Py4JJavaError: An error occurred while calling o714.saveAsTextFile.
: org.apache.hadoop.mapred.FileAlreadyExistsException: Output directory 
XXXXXXXX/user/myname/all.txt already exists

为了避免混淆，我将 IP 地址“模糊”为 XXXXXXXX。

这样做的正确方法是什么？我将拥有大量数据集(例如“text1”、“text2”...)，并且希望在将它们保存到 HDFS 之前对每个数据集执行 python 函数。但我希望将结果全部放在“一个”输出文件中。

非常感谢!
MG

编辑:看来我的最终目标还不是很明确。我需要分别对每个文本文件应用一个函数，然后我想将输出附加到现有的输出目录。像这样的事情:

for i in list:
    text_file = sc.textFile("/user/myname/" + i)
    text_file = really_cool_python_function(text_file)
    text_file.saveAsTextFile("/user/myname/all.txt")

Answer 1

我想将此作为评论发布，但由于我没有足够的声誉而无法这样做。

您必须将 RDD 转换为数据帧，然后以追加模式写入。要将 RDD 转换为数据帧，请查看以下答案:
https://stackoverflow.com/a/39705464/3287419
或此链接 http://spark.apache.org/docs/latest/sql-programming-guide.html
要以追加模式保存数据帧，以下链接可能有用:
http://spark.apache.org/docs/latest/sql-programming-guide.html#save-modes

几乎同样的问题也在这里Spark: Saving RDD in an already existing path in HDFS 。但提供的答案是针对 scala 的。我希望 python 也能完成类似的事情。

还有另一种(但丑陋的)方法。将 RDD 转换为字符串。令结果字符串为 resultString 。使用子进程将该字符串附加到目标文件，即

subprocess.call("echo "+resultString+" | hdfs dfs -appendToFile - <destination>", shell=True)