问题标签 [flink-batch]

下面的代码在本地运行，但不在集群上。它挂在 GroupReduceFunction 上，即使在几小时后也不会终止（大数据需要 9 分钟才能在本地计算）。日志中的最后一条消息：

GroupReduce (GroupReduce at main(MyClass.java:80)) (1/1) (...) switched from DEPLOYING to RUNNING.

代码片段：

有什么想法可以并行运行这个部分吗？提前致谢！

我在 Apache Mesos 1.8.1 上使用 Apache Flink 1.9.1，在 3 节点集群上使用或多或少的标准设置。第一个节点（192.168.10.11）运行 Mesos master 和 Flink App Master。其他两个节点（192.168.10.13、192.168.10.14）在某种程度上都是运行 mesos 代理并可以访问 Flink Worker docker 容器的从节点。Flink 在 Docker 中运行，但 Mesos Master 和代理在主机上本地运行。

一开始一切正常。我提交给 Flink 的第一个批处理作业被 App Master 正确处理，资源管理器与 Mesos 成功协商，资源管理器框架在其中一个从节点上分配了一个新的工作任务。任务成功完成，客户端程序得到它的结果。

几分钟后，ResourceManager 超时，Flink Worker Task（Docker 容器）因空闲超时而被杀死。到目前为止，一切都很好。

但是在那之后，如果我以与第一次相同的方式再次提交相同的作业，则资源管理器什么也不做，也不会尝试创建任何新的工作人员。作业等待 5 分钟，超时后失败。

如果有人能指出我可能出了什么问题，我会很高兴的。

下面是 Flink App Master 和 mesos master 相关日志的摘录：

1. Flink App Master 的启动

2. Flink App Master 启动后 mesos-master 的日志

3. 在第一次提交作业时登录 Flink App Master（工作正常）

4. mesos中对应的时间：

5. 几分钟后，TaskManager 的空闲超时开始，TaskManager 被杀死

6. mesos中对应的日志：

7. App Master 中的第二个作业提交（保持挂起并在超时后停止且未进行任何处理的作业）

在这个阶段，日志会停留几分钟。Mesos 中没有任何活动。

在我看来，问题显然出在 Flink App Master / Resource Manager 中，因为在 mesos 中再次发生了注意事项。在那一刻，mesos 主日志中没有任何活动（Mesos 控制台没有显示任何内容，没有提交或执行新任务，两个代理都没有做任何事情）。

几分钟后发生超时，作业在 FAILED 状态下停止。

8. 超时后第二个作业失败

作为记录，我提交的作业是来自http://diegoreico.com/environments/runningflinkgclusterwithzeppelin/的简单 Batch Hello World

更新 2019-10-24 - 下午 1:00

以下具有 DEBUG 级别的日志提取可用：

• Flink App master 启动：http ://www.niceideas.ch/flink/1_flink_app_master_startup.txt
• 第一个作业执行（成功）：https ://www.niceideas.ch/flink/2_first_job_execution_successful.txt
• 任务管理器空闲超时杀死：https ://www.niceideas.ch/flink/3_task_manager_idle_kill.txt
• 第二个作业执行（失败）：https ://www.niceideas.ch/flink/4_second_job_execution_failed.txt

更新 2019-10-24 - 下午 5:30

• 任务管理器（Flink worker）的日志可以在这里找到：https ://www.niceideas.ch/flink/5_task_manger_logs.txt

我正在使用DataSetAPI 使用 Apache Flink 编写批处理作业。我可以使用读取文本文件，readTextFile()但此功能一次只能读取一个文件。

如果可能的话，我希望能够一个一个地使用我目录中的所有文本文件并一个一个地同时处理它们，在与DataSetAPI 批处理作业相同的功能中。

另一个选项是实现一个循环执行多个作业，每个文件一个，而不是一个作业，多个文件。但我认为这个解决方案不是最好的。

有什么建议吗？

我有一个应用程序从流中接收大部分输入，但它的一些数据来自 RDBMS 和一系列静态文件。

流将不断发出事件，因此 flink 作业永远不会结束，但是如何定期刷新 RDBMS 数据和静态文件以捕获对这些源的任何更新？

我目前正在使用JDBCInputFormat从数据库中读取数据。

下面是我正在尝试做的粗略示意图：

Flink 是否支持 Dataset(Batch Api) 中的 Side Outputs 功能？如果不是，从文件加载时如何处理有效和无效记录？

Flink：如何在 Flink 中处理 Null 值，尤其是在读取 CSV 等文件时。

我来自火花背景。所以火花中有空处理。像 df.na.fill 等

所以我想知道 Dataset API 中是否有这样的工具，或者唯一的方法是使用 map 函数。

使用 POJO 而不是 Case Class 更好吗？POJO中启动的默认值是否会在数据文件中缺少该值时生效。

我正在使用 flink AvroOutputFormat 写一些数据，

这会将数据写入一个名为test_1.avro. 当我试图读取文件时，

这会将行打印为，

是否有可能打印此数据值而不是对象地址的方法。

我想知道是否无论如何都使用 Flink 的数据流 API 从传入的记录中删除重复项（可能在特定时间窗口内），就像在提供称为“不同”的转换的 Dataset API 中一样。或者无论如何，如果数据集可以转换为数据流，假设数据集被转换为数据流以在 Flink 中进行内部处理。

请帮助我。提前致谢！干杯!

运行此命令时出现问题

我正在使用 VM，我的集群由 3 个任务管理器组成，主服务器也是作业管理器和任务管理器（4 个任务管理器和一个作业管理器）

部分例外

但是日志没有写入日志文件（ps：其他日志输出效果很好）。我认为 env.execute() 正在阻塞调用，并且当所有子任务都结束时 getAccumulatorResult() 执行。我不知道为什么最后一行日志输出不起作用。