Flink StateFun совмещенные функции связи - PullRequest
Новая
       75

Flink StateFun совмещенные функции связи

1 голос
/ 03 мая 2020

У меня правильно работает встроенное задание, и я хочу развернуть дополнительные совместные задания. Эти новые добавленные задания будут получать сообщения от старого задания и отправлять его на код kafka topi c.

, как показано ниже 

@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 -модуль

Вновь определенные модули будут получать сообщения от другого модуля и отправлять их kafka.

  • Нужно ли определять вход для каждой совмещенной работы? Если нет, то как я могу заставить эту работу работать?
  • Как я могу получить совместную работу для общения? Достаточно ли использовать один и тот же FunctionType?
  • Обмениваются ли совмещенные функции через вход / выход?

1 Ответ

2 голосов
/ 04 мая 2020

Ответы встроены, и к вашему сведению ничего, что вы спрашиваете, не находится в совместном месте, определяя c. Эти свойства сохраняются для удаленных модулей и заданий, которые содержат смешанные рабочие нагрузки, расположенные совместно и удаленно.

Нужно ли определять вход для каждой совмещенной работы? Если нет, то как я могу заставить эту работу работать?

Да, для каждой работы (удаленной или совместной) требуется как минимум один вход. Вход - это канал, который принимает сообщения из внешнего мира в приложение Statefun. Вспомните Кафку или Кинезис. Без входа задание никогда бы ничего не делало, потому что не было бы начальных сообщений, чтобы начать обработку.

Для каждого входа вы будете связывать 1 или более маршрутизаторов, которые берут каждое сообщение из входа и пересылают их до 0 или более функций на основе их типов функций [1].

Как получить совместную работу для общения? Достаточно ли использовать один и тот же FunctionType?

Да, функции просто сообщают друг другу, используя свои типы функций.

Связаны ли совместно расположенные функции через вход / выход?

Нет, сообщения передаются между функциями с использованием среды выполнения Apache 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

...