Funciones con estado de Apache Flink

Stateful Functions es una API que simplifica la creación de aplicaciones con estado distribuidas. Se basa en funciones con un estado persistente que pueden interactuar de forma dinámica con sólidas garantías de coherencia.

Una aplicación Stateful Functions es básicamente una aplicación de Apache Flink y, por lo tanto, se puede implementar en Managed Service para Apache Flink. Sin embargo, hay un par de diferencias entre el empaquetado de Stateful Functions para un clúster de Kubernetes y el de Managed Service para Apache Flink. El aspecto más importante de una aplicación Stateful Functions es que la configuración del módulo contiene toda la información de tiempo de ejecución necesaria para configurar el tiempo de ejecución de Stateful Functions. Por lo general, esta configuración se empaqueta en un contenedor específico de Stateful Functions y se implementa en Kubernetes. Pero eso no es posible con Managed Service para Apache Flink.

A continuación se presenta una adaptación del ejemplo de StateFun Python para Managed Service for Apache Flink:

Plantilla de aplicación Apache Flink

En lugar de utilizar un contenedor de cliente para el tiempo de ejecución de Stateful Functions, los clientes pueden compilar un contenedor de aplicaciones de Flink que simplemente invoque el tiempo de ejecución de Stateful Functions y contenga las dependencias necesarias. En el caso de Flink 1.13, las dependencias requeridas son similares a las siguientes:

<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>

Y el método principal de la aplicación Flink para invocar el tiempo de ejecución de Stateful Function es el siguiente:

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);
}

Tenga en cuenta que estos componentes son genéricos e independientes de la lógica que se implementa en Stateful Function.

Ubicación de la configuración del módulo

La configuración del módulo Stateful Functions debe incluirse en la ruta de la clase para que pueda detectarse en el tiempo de ejecución de Stateful Functions. Es mejor incluirlo en la carpeta de recursos de la aplicación Flink y empaquetarlo en el archivo jar.

De forma similar a una aplicación común de Apache Flink, puede usar Maven para crear un archivo uber jar e implementarlo en Managed Service para Apache Flink.