1

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

代码如下

@AutoService(StatefulFunctionModule.class)
public final class CoLocatedModule implements StatefulFunctionModule {

  @Override
  public void configure(Map<String, String> globalConfiguration, Binder binder) {
    FunctionProvider provider = new FunctionProvider();
    binder.bindFunctionProvider( CoLocated.TYPE,provider );

    binder.bindEgress(KafkaSpecs.TO_TRANSACTION_SPEC);
  }
}

我收到如下错误

 The program finished with the following exception:

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: There are no ingress defined.
    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:335)
    at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205)
    at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:138)
    at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:664)
    at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:213)
    at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:895)
    at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:968)
    at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30)
    at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:968)
Caused by: java.lang.IllegalStateException: There are no ingress defined.
    at org.apache.flink.statefun.flink.core.StatefulFunctionsUniverseValidator.validate(StatefulFunctionsUniverseValidator.java:25)
    at org.apache.flink.statefun.flink.core.StatefulFunctionsJob.main(StatefulFunctionsJob.java:66)
    at org.apache.flink.statefun.flink.core.StatefulFunctionsJob.main(StatefulFunctionsJob.java:41)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:567)
    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:321)

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

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

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

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

1 回答 1

3

内联响应，仅供参考，您所要求的任何内容都不是特定的。这些属性适用于包含协同工作和远程混合工作负载的远程模块和作业。

我是否必须为每个同地工作定义入口？如果不是，我怎样才能使这项工作？

是的，每项工作（远程或托管）都需要至少一个入口。入口是一个将来自外部世界的消息消费到 statefun 应用程序中的通道。想想 Kafka 或 Kinesis。如果没有入口，该作业将永远不会做任何事情，因为将没有初始消息来开始处理。

对于每个入口，您将绑定 1 个或多个路由器，这些路由器从入口获取每条消息并根据其功能类型将它们转发到 0 个或多个功能[1]。

我怎样才能让同地工作进行交流？使用相同的 FunctionType 就足够了吗？

是的，函数只是使用它们的函数类型相互发送消息。

位于同一地点的功能是否通过入口/出口进行通信？

不，消息使用 Apache Flink 运行时在函数之间传递，其中包含高度优化的网络堆栈。一旦从入口中提取消息，它就不再与该入口交互。如果有兴趣，您可以在社区撰写的一些博客文章中了解 Flink 的网络堆栈是如何工作的，但这并不是在生产中成功使用 statefun 的必要条件[2]。

[1] https://ci.apache.org/projects/flink/flink-statefun-docs-release-2.0/io-module/index.html#router

[2] https://flink.apache.org/2019/06/05/flink-network-stack.html

于 2020-05-04T16:19:13.660 回答