apache-flink - Flink 批处理 : data local planning on HDFS?

Question

我们一直在玩 Flink。到目前为止，我们一直在 Hadoop 2.x/YARN 上使用 Spark 和标准 M/R。

除了 YARN 上的 Flink 执行模型之外，AFAIK 不像 spark 那样动态，执行器在 YARN 中动态获取和释放虚拟核心，问题的要点如下。

Flink 看起来很神奇:用于流式 API ，我只会说它很棒而且很出色。

批处理 API:处理图非常强大，并且以独特的方式进行了优化和并行运行，比 Spark 和其他人更能利用集群可扩展性，优化了共享通用处理步骤的非常复杂的 DAG。

我发现的唯一缺点，我希望这只是我的误解和缺乏知识，是在规划使用 HDFS 上的输入的批处理作业时，它似乎不喜欢数据本地处理。

不幸的是，这不是一个小问题，因为在 90% 的用例中，您在 HDFS 上有一个大数据分区存储，通常您会执行以下操作:

读取和过滤(例如，只获取失败或成功)

聚合，减少，使用它

第一部分，当在简单的 M/R 或 spark 中完成时，总是使用 ' 的习语来计划。首选本地处理 '，因此数据由保存数据 block 的同一节点处理，速度更快，以避免通过网络传输数据。

在我们使用 3 个节点的集群进行的测试中，设置专门测试此功能和行为，Flink 似乎可以完美地处理 HDFS block ，例如如果文件由 3 个 block 组成，那么 Flink 可以完美地处理 3 个输入拆分并并行调度它们。
但是没有数据局部性模式。

请分享您的意见，我希望我只是错过了一些东西，或者它可能已经出现在新版本中。
提前感谢任何花时间回答这个问题的人。

Answer 1

Flink 使用与 Hadoop 和 Spark 不同的本地输入拆分处理方法。 Hadoop 为每个输入拆分创建一个 Map 任务，该任务最好调度到托管拆分引用的数据的节点。

相比之下，Flink 使用固定数量的数据源任务，即数据源任务的数量取决于算子配置的并行度，而不是输入拆分的数量。这些数据源任务在集群中的某个节点上启动，并开始从主节点(JobManager)请求输入拆分。在 HDFS 中文件的输入拆分的情况下，JobManager 分配具有位置首选项的输入拆分。所以有来自 HDFS 的本地感知读取。但是，如果并行任务的数量远低于 HDFS 节点的数量，则将远程读取许多拆分，因为源任务保留在启动它们的节点上，并且一个接一个地获取一个拆分(首先是本地任务，以后有远程的)。如果您的拆分非常小，也可能发生竞争条件，因为第一个数据源任务可能会在其他源任务执行第一个请求之前快速请求并处理所有拆分。

IIRC，本地和远程输入拆分分配的数量被写入 JobManager 日志文件，也可能显示在 Web 仪表板中。这可能有助于进一步调试问题。如果您发现问题似乎与我上面解释的不匹配，如果您可以通过用户邮件列表与 Flink 社区联系以找出问题所在，那就太好了。