Funções com estado do Apache Flink

Stateful Functions é uma API que simplifica a criação de aplicativos distribuídos com estado. É baseado em funções com estado persistente que podem interagir dinamicamente com fortes garantias de consistência.

Um aplicativo Stateful Functions é basicamente um aplicativo Apache Flink e, portanto, pode ser implantado no Managed Service for Apache Flink. No entanto, há algumas diferenças entre empacotar Stateful Functions para um cluster Kubernetes e para o Managed Service for Apache Flink. O aspecto mais importante de um aplicativo Stateful Functions é que a configuração do módulo contém todas as informações de runtime necessárias para configurar o runtime do Stateful Functions. Essa configuração geralmente é empacotada em um contêiner específico do Stateful Functions e implantada no Kubernetes. Mas isso não é possível com o Managed Service for Apache Flink.

A seguir está uma adaptação do exemplo em StateFun Python para Managed Service para Apache Flink:

Modelo de aplicativo Apache Flink

Em vez de usar um contêiner de cliente para o runtime do Stateful Functions, os clientes podem compilar um jar de aplicativo Flink que apenas invoca o runtime do Stateful Functions e contém as dependências necessárias. Para o Flink 1.13, as dependências necessárias são semelhantes a estas:

<dependency>
 <groupId>org.apache.flink</groupId>
 <artifactId>statefun-flink-distribution</artifactId>
 <version>3.1.0</version>
 <exclusions>
  <exclusion>
   <groupId>org.slf4j</groupId>
   <artifactId>slf4j-log4j12</artifactId>
  </exclusion>
  <exclusion>
   <groupId>log4j</groupId>
   <artifactId>log4j</artifactId>
  </exclusion>
 </exclusions>
</dependency>

E o método principal do aplicativo Flink para invocar o runtime do Stateful Function é assim:

public static void main(String[] args) throws Exception {
 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

 StatefulFunctionsConfig stateFunConfig = StatefulFunctionsConfig.fromEnvironment(env);

 stateFunConfig.setProvider((StatefulFunctionsUniverseProvider) (classLoader, statefulFunctionsConfig) -> {
  Modules modules = Modules.loadFromClassPath();
  return modules.createStatefulFunctionsUniverse(stateFunConfig);
 });

 StatefulFunctionsJob.main(env, stateFunConfig);
}

Observe que esses componentes são genéricos e independentes da lógica implementada no Stateful Function.

A localização da configuração do módulo

A configuração do módulo Stateful Functions precisa ser incluída no caminho da classe para ser descoberta no runtime do Stateful Functions. É melhor incluí-lo na pasta de recursos do aplicativo Flink e empacotá-lo no arquivo jar.

Semelhante a um aplicativo comum do Apache Flink, você pode usar o Maven em seguida para criar um arquivo uber jar e implantá-lo no Managed Service for Apache Flink.