SDK for Flink DataStream Integration

This SDK may be used if you want your Stateful Functions application to consume events from, or output events to Flink DataStreams. Using this SDK, you may combine pipelines written with the Flink DataStream API or higher-level libraries (such as Table API, CEP etc., basically anything that produces a DataStream) with the programming constructs provided by Stateful Functions to build complex streaming applications.

To use this, add the Flink DataStream Integration SDK as a dependency to your application:

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>statefun-flink-datastream</artifactId>
  4.     <version>2.2.0</version>
  5. </dependency>

SDK Overview

The following sections covers the important parts on getting started with the SDK. For the full code and working example, please take a look at this example.

Preparing a DataStream Ingress

All DataStreams used as ingresses must contain stream elements of type RoutableMessage. Each RoutableMessage carries information about the target function’s address alongside the input event payload.

You can use the RoutableMessageBuilder to transform your DataStreams:

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  3. DataStream<String> names = env.addSource(...)
  5. DataStream<RoutableMessage> namesIngress =
  6.     names.map(name ->
  7.         RoutableMessageBuilder.builder()
  8.             .withTargetAddress(new FunctionType("example", "greet"), name)
  9.             .withMessageBody(name)
  10.             .build()
  11.     );

In the above example, we transformed a DataStream<String> into a DataStream<RoutableMessage> by mapping element in the original stream to a RoutableMessage, with each element targeted for the function type (example:greet).

Binding Functions, Ingresses, and Egresses

Once you have transformed your stream ingresses, you may start binding functions to consume the stream events, as well as DataStream egresses to produce the outputs to:

  1. FunctionType GREET = new FunctionType("example", "greet");
  2. FunctionType REMOTE_GREET = new FunctionType("example", "remote-greet");
  3. EgressIdentifier<String> GREETINGS = new EgressIdentifier<>("example", "greetings", String.class);
  5. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  7. DataStream<String> names = env.addSource(...);
  9. DataStream<RoutableMessage> namesIngress =
  10.     names.map(name ->
  11.         RoutableMessageBuilder.builder()
  12.             .withTargetAddress(GREET, name)
  13.             .withMessageBody(name)
  14.             .build()
  15.     );
  17. StatefulFunctionEgressStreams egresses =
  18.     StatefulFunctionDataStreamBuilder.builder("example")
  19.         .withDataStreamAsIngress(namesIngress)
  20.         .withRequestReplyRemoteFunction(
  21.             RequestReplyFunctionBuilder.requestReplyFunctionBuilder(
  22.                     REMOTE_GREET, URI.create("http://localhost:5000/statefun"))
  23.                 .withPersistedState("seen_count")
  24.                 .withMaxRequestDuration(Duration.ofSeconds(15))
  25.                 .withMaxNumBatchRequests(500))
  26.         .withFunctionProvider(GREET, unused -> new MyFunction())
  27.         .withEgressId(GREETINGS)
  28.         .build(env);

As you can see, instead of binding functions, ingresses, and egresses through modules as you would with a typical Stateful Functions application, you bind them directly to the DataStream job using a StatefulFunctionDataStreamBuilder:

  • Remote functions are bound using the withRequestReplyRemoteFunction method. Specification of the remote function such as service endpoint and various connection configurations can be set using the provided RequestReplyFunctionBuilder.
  • Embedded functions are bind using withFunctionProvider.
  • Egress identifiers used by functions need to be bind with the withEgressId method.

Consuming a DataStream Egress

Finally, you can obtain an egress as a DataStream from the result StatefulFunctionEgressStreams:

  1. EgressIdentifier<String> GREETINGS = new EgressIdentifier<>("example", "greetings", String.class);
  3. StatefulFunctionEgressStreams egresses = ...
  5. DataStream<String> greetingsEgress = egresses.getDataStreamForEgressId(GREETINGS);

The obtained egress DataStream can be further processed as in a typical Flink streaming application.

Configuration

Like a typical Stateful Functions application, configuration specific to Stateful Functions is set through the flink-conf.yaml file, as explained here. You can also overwrite the base settings for each individual job:

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  3. StatefulFunctionsConfig statefunConfig = StatefulFunctionsConfig.fromEnvironment(env);
  4. statefunConfig.setGlobalConfiguration("someGlobalConfig", "foobar");
  5. statefunConfig.setFactoryType(MessageFactoryType.WITH_KRYO_PAYLOADS);
  7. StatefulFunctionEgressStreams egresses =
  8.     StatefulFunctionDataStreamBuilder.builder("example")
  9.         ...
  10.         .withConfiguration(statefunConfig)
  11.         .build(env);

Attention: The setFlinkJobName method on StatefulFunctionsConfig does not have effect using this SDK. You need to define the job name as you normally would via Flink’s DataStream API.