apache-spark - Spark 并发写入相同的 HDFS 位置

Question

我有一个 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 处理器，但是我会失去性能。

需要您的专家建议。

提前致谢。

Answer 1

问题似乎是两个 Spark 节点独立地尝试写入同一个地方，导致冲突，因为最快的一个会在第二个期望它之前清除工作目录。
最直接的解决方案可能是避免这种情况。
据我了解您如何使用Nifi和spark，Nifi运行的节点也决定了spark运行的节点(存在1-1关系？)
如果是这种情况，您应该能够通过将 Nifi 中的工作路由到互不干扰的节点来解决此问题。查看依赖属性的负载均衡策略(队列的属性)。当然，您需要定义正确的属性，但诸如目录或表名之类的东西应该有很长的路要走。