问题标签 [flink-statefun]

我们有兴趣从新的 Stateful Functions连接到常规的 Flink Streaming 应用程序，最好使用 Table API。想法是从 Statefun 咨询在 Flink 中注册的表，这可能吗，正确的做法是什么？

到目前为止，我的想法是在一些主函数中初始化我的表流并注册一个有状态的函数提供程序以连接到表：

有状态函数提供者将返回 a FnTableQuery，它在收到消息时简单地查询表：

然后查询函数对象将作为每个已建立进程的参与者运行，并在调用时简单地查询表：

如果这种方法没有意义，我提前道歉，因为我不知道：

1. Flink 和 Statefun 应用程序可以在源/接收器领域之外一起工作，特别是因为这个特定的功能是无状态的并且表是有状态的

2. 我们可以像这样查询 Flink 表，我只是将它们作为中间对象进行查询以发送到接收器或数据流

3. 在 Module.configure 中初始化事物是有意义的，并且如果每个并行工作者都调用有状态函数提供程序及其匹配函数一次

我研究了 Apache Flink 的新 Stateful Functions 2.0 API。我阅读了以下文档链接https://ci.apache.org/projects/flink/flink-statefun-docs-stable/。我还在 Git 存储库中运行了示例。（https://github.com/ververica/stateful-functions/tree/master/stateful-functions-examples）关于实施我有几个问题。

https://flink.apache.org/stateful-functions.html --> 页面末尾有一个例子是欺诈检测的交易评分。

第一个问题是关于状态 TTL。我怎样才能把状态交给 TTL？示例说：30 天后，“欺诈计数”功能将收到一条过期消息（来自自身）并清除其状态。我应该做这本手册还是有其他功能？我该如何做这本手册？

关于keyedstream的第二个问题。示例说：将存在多个“欺诈计数”实例 - 例如，每个客户帐户一个。我应该把价值观PersistedTable<K,V>吗？例如<customerid,count>. 我可以清除特定键的状态吗？

最后一个问题是关于窗口和水印。如何将这些功能实现到 Stateful Functions 2.0？

Flink Stateful Functions 2.0 能够进行异步调用，例如对外部 API：[ https://ci.apache.org/projects/flink/flink-statefun-docs-release-2.0/sdk/java.html#完成异步请求][1]。

然后暂停函数执行，直到调用完成并显示 Success、Failure 或 Unknown。未知的是：

在 CompletableFuture 完成之前，有状态函数可能在另一台机器上重新启动，因此不知道异步操作的状态是什么。

当暂停/等待函数再次调用具有相同 ID 的调用时会发生什么？

1. 然后被调用者是否等待被调用函数对其异步结果的处理，以便第二次调用以干净、非共享的异步后状态执行？
2. 或者第二个调用是否按正常计划执行，因此在异步调用的当前状态之上，然后当异步调用完成时，它继续使用在异步调用挂起时更新的状态进行处理？
3. 或者调用算作被调用函数的“重启”——在这种情况下，执行顺序是什么：“重启”运行，然后异步返回“重启”以从现在更新的状态执行，或者这个顺序是颠倒的？
4. 或者是其他东西？

当我检查状态函数模块时，我发现了以下结果。

https://ci.apache.org/projects/flink/flink-statefun-docs-release-2.0/sdk/python.html -> 支持 Python SDK 作为远程模块。

https://ci.apache.org/projects/flink/flink-statefun-docs-release-2.0/sdk/java.html -> 支持 Java SDK 作为嵌入式模块。

https://ci.apache.org/projects/flink/flink-statefun-docs-release-2.0/sdk/modules.html -> 此模块类型可以支持任意数量的语言 SDK。远程模块通过 YAML 配置文件向系统注册。

Stateful Function 应用程序使用 http 与 Apache Flink 运行时通信。Python SDK 附带一个 RequestReplyHandler，它会根据 RESTful HTTP POSTS 自动分派函数调用。RequestReplyHandler 可以使用任何 HTTP 框架公开。

如何在 JAVA 中处理请求？是否有 Java SDK 的示例远程功能？

我还可以混合嵌入式和同位/远程功能吗？例如，如果我将客户的统计信息（总和、计数、平均等）输出到另一个 kafka 主题，然后我进入该主题，我可以使用 python sdk 的远程函数。远程功能包括可更新的场景，但嵌入式功能是客户的核心统计数据，无需更新这些功能。

我正在尝试使用 apache flink 有状态函数来实现消息传递场景。

按照设计，我需要从传入消息中计算一些统计数据并将它们存储在状态中。之后，场景函数将访问这些状态和消息并在它们上运行业务规则。但是我们每条消息可能有几十个场景，每个场景都应该只运行一次。

代码或多或少如下

FnScenarioManager：

我的问题是如果集群在runTransactionScenarios中间崩溃会发生什么？

• 每个场景会只运行一次吗？如果不是，我怎么能确保呢？

重新部署 Flink 有状态函数时，我找不到任何关于版本控制的问题。

当我为当前正在执行的实例的有状态函数更新和部署代码时，这会对以下内容产生什么影响：

1. 当前正在执行的实例，而不是等待回调？
2. 当前正在等待异步回调的实例（回调返回时运行什么代码）？
3. 如果其中一个或 1 或 2 可能正在执行新代码，如何处理旧版本和新执行版本之间 2 个函数版本的可能不同的持久状态？
4. 如果 2 没有自动升级，当我希望长时间运行的函数发生这种情况时，如何使新函数版本运行（比如它正在等待 30 天计时器上的自我回调，我希望新版本在回调返回）？如果我可以升级版本，处理版本之间可能添加或删除持久状态的方法是什么？

我有一个正常工作的嵌入式工作，我想部署额外的同地工作。这些新添加的作业将接收来自旧作业的消息并将其发送到 kafka 主题。

代码如下

我收到如下错误

该错误非常不言自明，希望我定义入口。

链接中有一个类似定义的嵌入式模块-> https://ci.apache.org/projects/flink/flink-statefun-docs-stable/sdk/modules.html#embedded-module

新定义的模块将接收来自另一个模块的消息并将它们发送给 kafka。

• 我是否必须为每个同地工作定义入口？如果不是，我怎样才能使这项工作？
• 我怎样才能让同地工作进行交流？使用相同的 FunctionType 就足够了吗？
• 位于同一地点的功能是否通过入口/出口进行通信？

我曾计划用 python statefun 示例交流嵌入式有状态函数（链接：https ://github.com/apache/flink-statefun/tree/master/statefun-examples/statefun-python-greeter-example ）

但是（总是有一个但是）我一直在努力构建 python 发行版。我认为我有正确的配置。

这是我试图描述的错误

我的点子版本：19.2.3

我的python版本：3.7.5

顺便说一下 python sdk setup 说的；

它说 3.7 python 没问题。

你有什么想法？

谢谢。

我正在尝试将传入消息发送到多个有状态函数，但我无法完全理解该怎么做。为了便于理解，假设我的一个有状态函数获取一些整数并将它们发送到几个远程函数。这些函数将此整数添加到它们的状态值中，并将其保存为新状态。

当这两个远程功能之一失败时，另一个应该继续以相同的方式工作。当失败的函数恢复时，它应该处理它在失败期间无法处理的消息。

我想像下面这样一个接一个地发送它们，但我认为它不会起作用

• 我怎样才能以容错的方式做到这一点？
• 如果可能的话，它如何在后台工作？

开发apache flink stateful function时使用python或者java有什么优缺点。

• 有性能差异吗？对于相同的操作，哪一个更有效？
• 我们可以完全在 python 上开发应用程序吗？
• 一个支持而另一个不支持的功能是什么。