Stateful-Funktionen von Apache Flink

Stateful Functions ist eine API, die die Erstellung verteilter zustandsbehafteter Anwendungen vereinfacht. Sie basiert auf Funktionen mit persistentem Status, die dynamisch mit starken Konsistenzgarantien interagieren können.

Eine Stateful-Functions-Anwendung ist im Grunde lediglich eine Apache-Flink-Anwendung und kann daher in Managed Service für Apache Flink bereitgestellt werden. Es gibt jedoch einige Unterschiede zwischen der Paketierung von Stateful Functions für einen Kubernetes-Cluster und für Managed Service für Apache Flink. Der wichtigste Aspekt einer Stateful-Functions-Anwendung ist, dass die Modulkonfiguration alle erforderlichen Laufzeitinformationen zur Konfiguration der Stateful-Functions-Laufzeit enthält. Diese Konfiguration wird normalerweise in einen Stateful-Functions-spezifischen Container gepackt und auf Kubernetes bereitgestellt. Aber das ist mit Managed Service für Apache Flink nicht möglich.

Es folgt eine Anpassung des StateFun Python-Beispiels für Managed Service für Apache Flink:

Apache Flink-Anwendungsvorlage

Anstatt einen Kundencontainer für die Stateful-Functions-Laufzeit zu verwenden, können Kunden eine Flink-Anwendungs-JAR-Datei kompilieren, die lediglich die Stateful-Functions-Laufzeit aufruft und die erforderlichen Abhängigkeiten enthält. Für Flink 1.13 sehen die erforderlichen Abhängigkeiten etwa wie folgt aus:

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

Und die Hauptmethode der Flink-Anwendung zum Aufrufen der Stateful-Function-Laufzeit sieht so aus:

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

Beachten Sie, dass diese Komponenten generisch und unabhängig von der Logik sind, die in der Stateful Function implementiert ist.

Ort der Modulkonfiguration

Die Konfiguration des Stateful-Functions-Moduls muss im Klassenpfad enthalten sein, damit sie für die Stateful-Functions-Laufzeit auffindbar ist. Am besten fügen Sie es in den Ressourcenordner der Flink-Anwendung ein und packen es in die JAR-Datei.

Ähnlich wie bei einer gewöhnlichen Apache-Flink-Anwendung können Sie dann Maven verwenden, um eine Uber-JAR-Datei zu erstellen und diese auf Managed Service für Apache Flink bereitzustellen.