igalshilman opened a new pull request #133:
URL: https://github.com/apache/flink-statefun/pull/133
# Add DataStream API interoperability
This PR adds an extension the stateful function SDK that allows embedding
stateful function applications into a data stream program.
This integration allows:
* defining `DataStream`s as StateFun ingresses
* binding one or more stateful functions
* binding remote `RequestReply` functions
* and finally obtaining StateFun egresses as `DataStream`s.
Here is a short example snippet of how to insert a StateFun pipeline into a
regular DataStream pipeline. ([full
example](statefun-examples/statefun-flink-datastream-example/src/main/java/org/apache/flink/statefun/examples/datastream/Example.java))
```
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> namesStream = ... ;
DataStream<RoutableMessage> names =
namesStream
.map(
name ->
RoutableMessageBuilder.builder()
.withTargetAddress(GREET, name)
.withMessageBody(name)
.build());
...
StatefulFunctionEgressStreams out =
StatefulFunctionDataStreamBuilder.builder("example")
.withDataStreamAsIngress(namesDataStream)
.withFunctionProvider(GREET, unused -> new MyFunction())
.withRequestReplyRemoteFunction(
requestReplyFunctionBuilder(
REMOTE_GREET,
URI.create("http://localhost:5000/statefun";))
.withPersistedState("seen_count")
.withMaxRequestDuration(Duration.ofSeconds(15))
.withMaxNumBatchRequests(500))
.withEgressId(GREETINGS)
.withConfiguration(statefunConfig)
.build(env);
...
```
- A `RoutableMessage` is the entry point to the StateFun pipeline.
- Then it is possible to register 1 or more `DataStream<RoutableMessage>` as
ingresses.
- StateFun would deliver the `payloads` associated with the
`RoutableMessage` to the appropriate stateful function instance.
- Egress ids has to be explicitly defined, and they can be collected as a
`DataStream` from `StatefulFunctionEgressStreams`.
----
The following is a short description of the changes (not in the same commit
order)
* Add a `statefun-flink/statefun-flink-datastream` for the new SDK.
* Added a new example under examples/
* Restructured the translation logic so that it can be used both from the
data stream api and the regular statefun sdk.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
