问题标签 [flink-batch]

我的要求是将 30 天的数据保存到流中，以便在任何一天进行处理。所以当 FLINK 应用程序启动的第一天，它将从数据库中获取 30 天的数据并合并到当前的流数据中。我的挑战是 - 管理 30 天的数据窗口。如果我创建滑动窗口 30 天，滑动时间为 1 天。就像是

在这种情况下，当添加历史数据的第一个元素时， process() 不会立即开始处理。我的假设是```a）默认情况下，第一个事件将是第一个窗口的一部分，并且可以立即进行处理。b) 第二天的工作将从窗口中删除最后 29 天的数据。我的假设对那段代码是否正确？感谢您对此的帮助。

在 TaskManagers(TM) 的 HA 上下文中，当一个 TM 发生故障时，一个新的 TM 将从 JobManager(JM) 故障的最新检查点恢复。

假设我们有 3 个 TM ( tm1, tm2, & tm3) 在某个时间t每个人的检查点 (cp) 都在cp1。所有 TM 在它们之间广播数据。

现在tm2下去了，JM 带来tm2'了cp1检查点作为 HA 的一部分。当t+x一个新的 TM 被提出时，同时其他人进展到cp2.

tm1由广播和tm3作为其一部分的数据如何cp2重播tm2'？

我的要求是围绕flink中的sql查询结果处理或构建一些逻辑。为简单起见，假设我有两个 sql 查询，它们在不同的窗口大小和一个事件流上运行。我的问题是

• a)我怎么知道这是哪个查询结果
• b）我怎么知道执行查询的结果有多少行？我需要此信息，因为我必须构建带有事件列表的通知消息，这些事件列表是查询结果的一部分。

感谢你的帮助。谢谢阿舒托什

我在 IntelliJ 中工作以创建和测试一个机器学习模型，该模型将对来自流的传入数据进行分类。我正在使用 Scala (2.11.8) 并使用 Flink 框架 (1.8.3)。我正在尝试读取一个 csv 文件，该文件包含我的训练数据，其中包含 long、double 和 boolean 数据类型的列，其中包含以下几行：

我得到的错误是：“未指定的值参数evidence$2。” 此参数对应于 TypeInformation 参数。

我需要进行哪些更改才能指定此参数？我的导入包括： import org.apache.flink.api.scala._ import org.apache.flink.streaming.api.scala._

附加信息： readCsvFile [T : ClassTag : TypeInformation](filePath: String): DataSet[T]

我们有一个来自数据库表的文件路径列表，其中包含创建时间的时间戳。试图弄清楚我们如何使用 db 中的文件路径列表仅将那些文件从 nfs 转发到 kafka sink。

现在，我正在使用带有文件夹根目录的 ContinuousFileMonitoringFunction 的自定义版本，该文件夹将包含 DB 将显示的所有文件。此操作非常缓慢，因为要遍历文件夹以收集有关更新文件的信息，因为该文件夹太大而只有几 TB 的数据。

ds 包含所有应该发送到 kafka 的文件路径。

以下是我计划实施的想法。但是考虑到 flink 并行性、flink 库支持等，有没有更有效的方法？

我正在尝试设置一个 flink-yarn 会话来运行约 100 多个批处理作业。在连接到约 40 个任务管理器和约 10 个正在运行的作业（每个任务管理器有 2 个插槽和每个 1GB 内存）之后，会话看起来变得不稳定。有足够的可用资源。flink UI 突然变得不可用了，我猜作业经理可能已经死了。最终，纱线应用程序也被杀死了。

作业管理器在 4 核 16GB 节点上运行 12GB 可用

是否有任何指南可以计算作业管理器资源与其可以处理的任务管理器数量？

我试图看看用 apache flink 作业（以及 kuberentes 作为底层基础）替换使用 SSIS 包创建的数百个提要文件 ETL 作业的可行性。我在一些文章中看到的一个建议是“将一个 flink 集群用于一种类型的工作”。

由于我每天每种作业类型都有少量作业，那么这意味着对我来说最好的方法是在执行作业时动态创建 flinkcluster 并销毁它以释放资源，这是正确的方法吗？我正在设置没有作业管理器的 flinkcluster。

关于使用 flink 进行批处理 ETL 活动的最佳实践的任何建议。

可能是最重要的问题：flink 是问题陈述的正确解决方案，还是我应该更多地研究 Talend 和其他经典 ETL 工具？

如何使用独立应用程序（本地执行环境或迷你集群）实现 flink 保存点。我在 flink-config.yaml 文件中配置了保存点目录，但不确定如何在关闭应用程序之前获取保存点以及如何通过重新启动应用程序来恢复？

有没有办法或必须使用 flink 集群然后使用 CLI。

感谢你的帮助。谢谢

我有一个 flink 批处理作业，它从 S3 读取一个非常大的镶木地板文件，然后它将一个 json 下沉到 Kafka 主题中。

问题是如何使文件读取过程有状态？我的意思是每当工作中断或崩溃时，工作应该从以前的阅读状态开始？我不想在作业重新启动时向 Kafka 发送重复的项目。

这是我的示例代码

我需要关于在 K8 上部署 Flink 应用程序的帮助

我们有 3 个源会以 SQL 查询的形式发送触发条件。总查询数约为 3-6k，实际上是 flink 实例的沉重负载。我尝试执行，但它非常慢并且需要很多时间才能开始。

由于查询量很大，我们决定为每个源创建多个 flink 应用实例。所以有效地一个 flink 实例将只执行 ~1-2K 查询。

示例：sql查询源为A、B、C

Flink 实例：

App A --> 将只负责处理源 A 查询

App B --> 将只负责处理源 B 查询

App C --> 将只负责处理源 C 查询

我想在 Kubernetes 上部署这些实例

问题：

a）是否可以使用迷你集群（内置）部署独立的 flink jar？就像启动主方法一样：Java -cp mainMethod（sourceName 是命令行参数 A/B/C）。

b) 如果 k8 的一个 pod 或 flink 实例宕机，那么我们如何在另一个 pod 或另一个 flink 实例中管理它？是否可以将工作交给其他 pod 或其他 flink 实例？

对不起，如果我把两个或更多的东西混在一起:(

感谢你的帮助。谢谢