问题标签 [flink-statefun]

我们能够随心所欲地扩展无状态函数。但是，如果我们在扩展无状态函数时不扩展它们，那么有状态函数可能会成为瓶颈。由于数据在节点之间的分布，缩放状态似乎很棘手。如果有状态函数成为瓶颈，我们也可以扩展它们吗？

如何动态更新 module.yaml？例如，我想添加新的远程功能，所以我必须更新 module.yaml 文件。我应该停止/启动 flink 集群以更新 module.yaml 吗？用例的真正方法是什么？更新或现有的远程功能作业很好。因为滚动更新就足够了。

我在 HA(Zookeeper) 模式下安装了 Flink Statefun Kubernetes Cluster(3 node)。该集群包括 statefun 2.0.0 镜像和嵌入式 jar、模块 yaml 文件。我还为远程功能安装了另一个 Kubernetes 集群。官方示例显示在同一个 Kubernetes 集群中。（我认为只是示例。通常部署远程函数另一个 k8s 集群会更好，因为可以扩展计算。）我应该为每个函数定义 LoadBalancer 服务吗？单LB多服务怎么样？具有多种服务的 Ingress 怎么样？ https://cloud.google.com/kubernetes-engine/docs/tutorials/http-balancer#step_6_optional_serve_multiple_applications_on_a_load_balancer

此外，我必须重新启动集群以更新 module.yaml，因此另一个 flink 集群可能是解决方案。例如嵌入式函数集群（包括 jar）和远程函数集群（包括 module.yaml）。我可以将这些集群部署到同一个 kubernetes 集群吗？

我正在尝试在 kubernetes 上部署高可用的 flink 集群。在下面的示例中，工作节点被复制，但我们只有一个主 pod。

https://github.com/apache/flink-statefun

据我了解，有两种方法可以使作业经理 HA。

1. https://ci.apache.org/projects/flink/flink-docs-stable/ops/jobmanager_high_availability.html
2. https://medium.com/hepsiburadatech/high-available-flink-cluster-on-kubernetes-setup-73b2baf9200e

在第一个示例中，我们部署另一个作业管理器以在发生故障时在它们之间切换。在第二个示例中，kubernetes 重新部署作业管理器 pod 以防发生故障

所以我有几个问题

• 对于这两个示例，当活动作业管理器失败时，正在运行的作业会发生什么情况？
• 第一个场景可以应用在 kubernetes 上吗？

• 对于作业管理器失败的第二种情况，flink UI 在 pod 恢复之前将不可用，但在第二种第一种情况下，它将可用，对吗？

• 这两种情况的优缺点是什么？

我正在使用flink stateful functions开发一个新项目。我已经使用FunctionTestHarness编写了一些基本的单元测试，但是使用此方法的测试无法测试有状态函数之间的交互。

flink测试文档（基本 flink，不适用于有状态函数）演示了如何使用作业的输出运行完整的作业MiniClusterWithClientResource，然后对作业的输出进行断言。我正在寻找一种与有状态函数类似的方法。

statefun -flink-harness-example看起来很有希望，但是RunnerTestusing Harness标记为 ，@Ignore因为它永远不会终止。这对于调试很有用，但不能用于自动化测试。

以下是我迄今为止发现的问题，这使得编写以Harness终止的测试变得困难：

1. Harness使用SerializableSupplier来提供输入，而SerializableSupplier没有办法说它已经完成。这意味着任何使用Harness的测试总是在等待更多的输入。
2. 如果Harness知道所有输入都已发送，那么一旦没有未决事件，它就需要一种终止方式。
3. 作为一个额外的复杂因素，由于Context发送的延迟事件，一些系统仍然永远不会终止.sendAfter()

我认为这将是实现更有趣的自动化测试的常见需求，这些自动化测试可以从 CI/CD 流程运行。有没有人找到解决上述问题的方法，或者使用Harness以外的工具发现了一种完全不同的方法？

我正在使用以下命令运行 stateful-fun 2.0 basic hello job

./bin/flink run -c org.apache.flink.statefun.flink.core.StatefulFunctionsJob ./stateful-sun-hello-java-1.0-SNAPSHOT-jar-with-dependencies.jar

我的 POM.xml 是

它给出了以下异常

该程序完成，但出现以下异常：

org.apache.flink.client.program.ProgramInvocationException：主要方法导致错误：无效配置：classloader.parent-first-patterns.additional；必须在 org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:335) 在 org.apache.flink 包含所有 org.apache.flink.statefun、org.apache.kafka、com.google.protobuf .client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205) 在 org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:138) 在 org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend .java:662) 在 org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210) 在 org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:893) 在 org. apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:

请建议如何解决这个问题。

我正在尝试从 flink statefun 退出到融合 kafka。在融合的 git repo 中，为了进行模式检查并将数据放入 kafka 主题，我们需要做的就是使用带有 avro 对象的 kafka 客户端 ProducerRecord 对象。

但是在statefun中，我们需要覆盖 kafka 出口的“ProducerRecord<byte[], byte[]> serialize”方法。这会导致以下错误。

Schema registery 和 statefun kafka egress 似乎不兼容。有什么解决方法吗？

我正在使用 Python SDK 试用 Stateful Functions 2.1 API，但我看不到如何在不阻塞应用程序的情况下对外部 api 进行异步调用的明确方法。

这是可能的还是有人可以让我走上正确的道路？

谁能告诉我如何在我的无服务器平台（FaaS）中设置 Flink 来执行事件驱动的操作？

我查看了 Flink 函数，它似乎很有希望。任何人都可以澄清以下内容吗？

1. 我需要在我的 FaaS 环境中安装什么。当事件（我的 s3 存储桶中的文件更改）发生时触发 flink 功能？

2. 我没有大数据平台，因此计划在我的 serverless/kubernetes 环境中使用 flink。

提前致谢！！

我正在尝试使用 apache flink 有状态函数来实现消息传递场景。我的状态之一能够通过提供给 MatchBinder 的两个不同函数进行更新。这两个函数基本上检查当前状态并相应地更新状态。

• 如果为同一个键同时调用这两个函数会发生什么？
• 是否有用于调用相同键的有状态函数的队列机制？
• 我们可以锁定状态访问/更新以进行顺序访问吗？