问题标签 [flink-statefun]

我们得到了例外，复制在这篇文章的末尾。提交新的 flink 作业时抛出异常；当 Flink 尝试恢复之前的状态时。

环境：

代码实现DataStream Transformation函数：ProcessFunction -> KeySelector -> ProcessFunction。入站消息按“sourceId”键进行分区，该键是异常堆栈跟踪的一部分。SourceId 是 String 类型，并且是唯一的。

我们已经覆盖了“org.apache.flink.streaming.api.functions.ProcessFunction.open()”方法任何帮助表示赞赏

异常堆栈跟踪：

我想使用早期触发逻辑进行窗口聚合（您可以认为聚合是由窗口关闭或特定事件触发的），我在文档上阅读：https://ci.apache。 org/projects/flink/flink-docs-release-1.12/dev/stream/operators/windows.html#incremental-window-aggregation-with-aggregatefunction

该文档提到Note that using ProcessWindowFunction for simple aggregates such as count is quite inefficient. 了这一点，因此建议与增量窗口聚合配对。

我的问题是文档中的AverageAggregate，状态没有保存在任何地方，所以如果应用程序崩溃，averageAggregate 会丢失所有中间值，对吗？

那么如果是这样的话，有没有办法做一个窗口聚合，仍然支持增量聚合，并且有一个状态后端从崩溃中恢复？

我是 flink 新手，正在寻找有关构建实时事件关联系统的建议。我有两个主要用例：

1. 事件关联逻辑由基于输入流中事件类型的静态规则组成。在最后 X 分钟内，根据这些规则关联不同事件类型的事件和具有业务价值的事件的输出数据。例如，在最后 1 分钟内，检查市场 A1 中事件类型 A 的价格是否 < 20,000，市场 M2 中事件类型 B 的价格是否小于 < 30,000，然后将事件 A 的数据添加到输出流，否则添加事件数据B.
2. 对于感兴趣/商业价值的事件，计算与最后 X 分钟的价格差异。例如，如果在应用所有规则后，我们决定事件 A 对最后 1 分钟窗口感兴趣，在将事件数据添加到输出流中之前，我们还想计算事件 A 与过去 10 分钟的价格差异。

为了实现这些用例，我正在通过输入数据中的产品类型 ID 在输入流上评估应用键。这将为我提供该产品针对不同市场的多种事件类型的数据，然后使用回溯期的滑动事件时间窗口，例如最后 10 分钟，滑动窗口为 1 分钟，并应用 ProcessWindowFunction 为最后 1 分钟的数据编写相关逻辑和使用其他 9 分钟的数据进行回顾并计算感兴趣事件的价格差异。

我不完全确定这是否是实现这些的最佳方式。任何提示/建议将不胜感激！

我们flink-conf.yaml的项目中有本地运行。我们希望能够在本地运行 flink 进行测试。我们团队的一部分人使用 Mac，另一部分人使用 PC。我们想设置state.checkpoint.dir一些普遍接受的路径，最好在用户的主目录下。问题是，是否可以根据我们正在运行的操作系统以编程方式设置它，如果没有，在此设置中是否有当前用户主目录的通用快捷方式，如 *NIX 上的“~”？

这是交易；

我正在处理在运行集群（flink statefun 2.2.1）时添加新的工作人员（嵌入式）。

如您所见，新的任务管理器可以注册到集群；

新部署的任务管理器的屏幕截图

但它没有初始化（它没有部署源）；

我在这里想念什么？（master和workers也必须相同的jar文件？或者使用jar文件部署taskmanager就足够了）

任何帮助将不胜感激，谢谢。

我从同一个 Kafka 主题入口处使用了多种消息类型。虽然我可以将消息类型扩展到“包装器”超类，然后在它们被反序列化后向下转换，但这似乎不是一个好的选择。有没有其他选择？

我有一个用例，我想使用 Flink 构建一个实时决策树评估器。我有一个类似于下面的决策树： 决策树示例

注意：这只是一个决策树的示例，我有多个这样的决策树，它们具有不同的产品类型/节点数和不同的条件。想写一个通用的 Flink 应用来评估所有这些。

现在在输入中，我每 1 分钟得到一个输入数据流，其中包含所有产品类型（A、B 和 c）的价格。为了实现我的用例，我能想到的一种方法如下：

1. 按产品类型过滤输入流
2. 对于每种产品类型，根据每分钟触发的产品类型，在过去 X 分钟内使用滑动窗口
3. 处理窗口功能，用于检查特定产品类型的价格差异，并在输出流中输出每种产品类型的价格差异。

现在我们有了树的每个产品类型/节点的价格差异，然后我们可以评估决策树逻辑。现在要做到这一点，我们必须确保在确定输出之前必须完成决策树中所有产品类型（上例中的产品 A、B 和 C）的价格差异计算处理。一种方法是将所有这些产品的输出从输出流存储到数据存储中，并每隔 5 秒左右从 ec2 实例检查所有这些价格计算是否完成。完成后，执行决策树逻辑以确定输出产品。

想了解是否有任何其他方式可以在 Flink 本身中完成整个计算而不需要任何其他组件（数据存储/EC2）。我对 Flink 还很陌生，所以任何线索都将受到高度赞赏！

我在 Apache Flink Statefun 3.0（自定义 Greeter 示例）中有一个嵌入式模块，它使用 JSON 序列化事件。当尝试route()从入口反序列化消息时，我收到一个异常，即我的自定义类型不能转换为 protobuf（是的，它不是）——但应该这样吗？我的意思是我尝试使用 3.x 文档，但没有发现关于要路由的类型的任何限制。

对此有任何提示或指示吗？

谢谢

我试图了解 Flink Statefun 模块的范围。假设我有一个第三方服务需要首先建立连接（例如凭证。这需要很长时间）然后，我可以与之交互。

我正在尝试了解 Statfun 模块的范围，以及是否应该为我的所有功能创建该连接，或者是否可以为每个模块创建该连接。

对于使用有状态函数的一些建议，我将不胜感激。我们目前正在使用 Flink，我们从多个 kafka 流中消费、聚合、运行计算，然后输出到新的流。问题是计算元素是由不同的团队提供的，他们选择的语言是 Python。我们希望为他们提供独立于流媒体元素开发和更新其组件的能力。最初，我们只是将他们的代码移植到 Java。有状态函数似乎在这里提供了一种替代方案，我们将保留我们的一些功能，并将模型作为 Python 中的有状态函数托管。但是，我想知道，与仅将计算模块托管在自己的管道上并在 Flink 中使用 AsyncFunction 与之交互相比，这是否有任何优势。