使用（spark、python、pyspark、jupyter）将多个项目保存到 HDFS答案

【问题标题】：Saving multiple items to HDFS with (spark, python, pyspark, jupyter)使用（spark、python、pyspark、jupyter）将多个项目保存到 HDFS
【发布时间】：2017-10-10 12:06:36
【问题描述】：

我习惯用 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")

【问题讨论】：

标签： python apache-spark hdfs pyspark-sql

【解决方案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)

【讨论】：

【解决方案2】：

您可以读取多个文件并保存它们

textfile = sc.textFile(','.join(['/user/myname/'+f for f in list]))
textfile.saveAsTextFile('/user/myname/all')

您将获得输出目录中的所有零件文件。

【讨论】：

看来我的最终目标不是很明确。我需要分别对每个文本文件应用一个函数，然后我想将输出附加到现有的输出目录。见编辑
所有文本文件的功能相同？
是的，所有文件都具有相同的功能，但我之前无法加入文本文件，因为每个文件都需要单独处理
所有文件中的列是相似的还是不同的？？

【解决方案3】：

如果文本文件都具有相同的架构，您可以使用Hive 将整个文件夹读取为单个表，并直接写入该输出。

【讨论】：

【解决方案4】：

我会试试这个，应该没问题：

   list = ['text1.txt', 'text2.txt', 'text3.txt', 'text4.txt']
    
    for i in list:
        text_file = sc.textFile("/user/myname/" + i)
    text_file.saveAsTextFile(f"/user/myname/{i}")

【讨论】：