Spark 并发写入相同的 HDFS 位置

Question

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

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

我正在使用 NiFi 集群上游的数据。NiFi 集群中的每个节点都会为消费数据创建一个流文件。我的火花代码正在处理该流文件。由于 NiFi 是分布式的，我的火花代码从不同的 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.

以下是问题/问题：

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

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

需要您的专家建议。

提前致谢。

Answer 1

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

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

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

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