在同一 HDFS 位置上触发并发写入答案

【问题标题】：Spark concurrent writes on same HDFS location在同一 HDFS 位置上触发并发写入
【发布时间】：2020-01-06 15:59:18
【问题描述】：

我有一个 spark 代码，它使用附加模式以 Json 格式将数据帧保存到 HDFS 位置（日期分区位置）。

df.write.mode("append").format('json').save(hdfsPath)
sample hdfs location : /tmp/table1/datepart=20190903

我正在使用 NiFi 集群上游的数据。 NiFi 集群中的每个节点都会为消费数据创建一个流文件。我的 spark 代码正在处理该流文件。随着 NiFi 的分发，我的 spark 代码从不同的 NiFi 节点并行执行，试图将数据保存到同一个 HDFS 位置。

由于我的数据是按日期分区的，因此我无法将 spark 作业的输出存储在不同的目录中。

这个过程从过去 14 天开始每天运行一次，我的 spark 作业失败了 4 次，出现不同的错误。第一个错误：

java.io.IOException: Failed to rename FileStatus{path=hdfs://tmp/table1/datepart=20190824/_temporary/0/task_20190824020604_0000_m_000000/part-00000-101aa2e2-85da-4067-9769-b4f6f6b8f276-c000.json; isDirectory=false; length=0; replication=3; blocksize=268435456; modification_time=1566630365451; access_time=1566630365034; owner=hive; group=hive; permission=rwxrwx--x; isSymlink=false} to hdfs://tmp/table1/datepart=20190824/part-00000-101aa2e2-85da-4067-9769-b4f6f6b8f276-c000.json

第二个错误：

java.io.FileNotFoundException: File hdfs://tmp/table1/datepart=20190825/_temporary/0 does not exist.

第三个错误：

java.io.FileNotFoundException: File hdfs://tmp/table1/datepart=20190901/_temporary/0/task_20190901020450_0000_m_000000 does not exist.

第四个错误：

java.io.FileNotFoundException: File hdfs://tmp/table1/datepart=20190903/_temporary/0 does not exist.

以下是问题/问题：

我无法再次重现此场景。该怎么做？
在所有 4 次中，错误都与 _temporary 目录有关。是因为 2 个或更多作业同时尝试将数据保存在同一个 HDFS 位置，而在执行该作业时，作业 A 可能已删除作业 B 的 _temporary 目录？（因为位置相同，所有文件夹都有共同的名称/_directory/0/）

如果是并发问题，那么我可以从主节点运行所有 NiFi 处理器，但我会失去性能。

需要您的专家建议。

提前致谢。

【问题讨论】：

第 2 点是问题所在。众所周知的事实。
@thebluephantom 你能分享一些关于这个问题的信息和链接吗？我尝试使用 Spark 并发进行搜索，但没有找到任何有用的信息。感谢您的快速回复。
稍后再做...
这不是我从上面假设的 S3
找不到信息。明天再看

标签： apache-spark hadoop apache-spark-sql hdfs apache-nifi

【解决方案1】：

问题似乎是两个 spark 节点独立地尝试写入同一个地方，从而导致冲突，因为最快的节点会在第二个节点期望它之前清除工作目录。

最直接的解决方案可能是避免这种情况。

据我了解你如何使用Nifi和spark，Nifi运行的节点也决定了spark运行的节点（有1-1的关系？）

如果是这种情况，您应该能够通过将 Nifi 中的工作路由到不相互干扰的节点来解决此问题。查看依赖于属性的负载均衡策略（队列的属性）。当然你需要定义正确的属性，但是像目录或表名这样的东西应该有很长的路要走。

【讨论】：