HDFS 文件接收器输出作为另一个流的文件流输入 - 竞争条件？答案

【问题标题】：HDFS file sink output as file stream input for another stream - race condition?HDFS 文件接收器输出作为另一个流的文件流输入 - 竞争条件？
【发布时间】：2018-12-21 15:20:39
【问题描述】：

我正在使用结构化流评估 15 节点 Spark 集群中的特定数据流。我在我的应用程序中定义了 2 个流式查询：

SQ1 - 从 Kakfa 读取数据 -> 进程 -> 写入 HDFS 文件接收器（路径 - hdfs://tmp/output）
SQ2 - 从 HDFS 以文件流的形式读取数据（与上述路径相同）-> 进一步处理 -> 使用 ForeachWriter 写入外部数据库

两个查询都设置为每 15 秒触发一次。

我的问题 - 我是否在这里查看竞争条件，其中 SQ2 从 HDFS 获取部分写入的文件（由 SQ1 生成）？一个更普遍的问题是，HDFS 的文件接收器编写器是“原子的”吗？我尝试在 Spark 中挖掘流式源代码，但没有取得太大进展。

【问题讨论】：

根据 Spark Streaming 文档here，似乎将文件写入 HDFS 是非原子的，即 SQ2 将拾取部分写入的文件。
具体来说，该文档指出：“完整”文件系统（如 HDFS）倾向于在创建输出流后立即对其文件设置修改时间。打开文件时，甚至在数据还没有已完全写入，它可能包含在 DStream 中 - 之后在同一窗口中对文件的更新将被忽略。即：可能会丢失更改，并且从流中省略数据。
你找到解决竞态条件的方法了吗？

标签： apache-spark hdfs spark-structured-streaming

【解决方案1】：

这种方式的主要问题是Spark Structured Streaming中的所有File Sink（比如HDFS）只能在append模式下运行。此外，一旦创建文件，就会从文件中读取。任何后续更新或写入完成都将被忽略。

根据“Learning Spark - 2nd Edition”一书关于从文件中读取

“每个文件都必须以原子方式出现在目录列表中——也就是说，整个文件必须立即可供读取，一旦可用，则无法更新或修改文件。”

"[Writing to files] ...它只支持追加模式，因为在输出目录中写入新文件（即追加数据到目录）很容易，但很难修改现有数据文件（正如更新和完整模式所期望的那样）。”

要克服您面临的问题，您可以将流式查询更改为：

SQ1 - 从 Kafka 读取数据 -> 进程 -> 进一步处理 -> 缓存/持久化
SQ2a 将缓存的 DataFrame 写入 HDFS 文件接收器（路径 - hdfs://tmp/output）
SQ2b 使用 ForeachWriter 将缓存的 DataFrame 写入外部数据库 -> 使用 ForeachWriter 写入外部数据库

【讨论】：