I/O Module

Stateful Functions’ I/O modules allow functions to receive and send messages to external systems. Based on the concept of Ingress (input) and Egress (output) points, and built on top of the Apache Flink® connector ecosystem, I/O modules enable functions to interact with the outside world through the style of message passing.

Ingress

An Ingress is an input point where data is consumed from an external system and forwarded to zero or more functions. It is defined via an IngressIdentifier and an IngressSpec.

An ingress identifier, similar to a function type, uniquely identifies an ingress by specifying its input type, a namespace, and a name.

The spec defines the details of how to connect to the external system, which is specific to each individual I/O module. Each identifier-spec pair is bound to the system inside an stateful function module.

  1. package org.apache.flink.statefun.docs.io.ingress;
  3. import org.apache.flink.statefun.docs.models.User;
  4. import org.apache.flink.statefun.sdk.io.IngressIdentifier;
  6. public final class Identifiers {
  8.     public static final IngressIdentifier<User> INGRESS =
  9.         new IngressIdentifier<>(User.class, "example", "user-ingress");
  10. }
  1. package org.apache.flink.statefun.docs.io.ingress;
  3. import java.util.Map;
  4. import org.apache.flink.statefun.docs.io.MissingImplementationException;
  5. import org.apache.flink.statefun.docs.models.User;
  6. import org.apache.flink.statefun.sdk.io.IngressIdentifier;
  7. import org.apache.flink.statefun.sdk.io.IngressSpec;
  8. import org.apache.flink.statefun.sdk.spi.StatefulFunctionModule;
  10. public class ModuleWithIngress implements StatefulFunctionModule {
  12.     @Override
  13.     public void configure(Map<String, String> globalConfiguration, Binder binder) {
  14.         IngressSpec<User> spec = createIngress(Identifiers.INGRESS);
  15.         binder.bindIngress(spec);
  16.     }
  18.     private IngressSpec<User> createIngress(IngressIdentifier<User> identifier) {
  19.         throw new MissingImplementationException("Replace with your specific ingress");
  20.     }
  21. }
  1. version: "1.0"
  3. module:
  4.      meta:
  5.          type: remote
  6.      spec:
  7.          ingresses:
  8.            - ingress:
  9.                meta:
  10.                  id: example/user-ingress
  11.                  type: # ingress type
  12.                spec: # ingress specific configurations

Router

A router is a stateless operator that takes each record from an ingress and routes it to zero or more functions. Routers are bound to the system via a stateful function module, and unlike other components, an ingress may have any number of routers.

  1. package org.apache.flink.statefun.docs.io.ingress;
  3. import org.apache.flink.statefun.docs.FnUser;
  4. import org.apache.flink.statefun.docs.models.User;
  5. import org.apache.flink.statefun.sdk.io.Router;
  7. public class UserRouter implements Router<User> {
  9.     @Override
  10.     public void route(User message, Downstream<User> downstream) {
  11.         downstream.forward(FnUser.TYPE, message.getUserId(), message);
  12.     }
  13. }
  1. package org.apache.flink.statefun.docs.io.ingress;
  3. import java.util.Map;
  4. import org.apache.flink.statefun.docs.models.User;
  5. import org.apache.flink.statefun.sdk.io.IngressIdentifier;
  6. import org.apache.flink.statefun.sdk.io.IngressSpec;
  7. import org.apache.flink.statefun.sdk.io.Router;
  8. import org.apache.flink.statefun.sdk.spi.StatefulFunctionModule;
  10. public class ModuleWithRouter implements StatefulFunctionModule {
  12.     @Override
  13.     public void configure(Map<String, String> globalConfiguration, Binder binder) {
  14.         IngressSpec<User> spec = createIngressSpec(Identifiers.INGRESS);
  15.         Router<User> router = new UserRouter();
  17.         binder.bindIngress(spec);
  18.         binder.bindIngressRouter(Identifiers.INGRESS, router);
  19.     }
  21.     private IngressSpec<User> createIngressSpec(IngressIdentifier<User> identifier) {
  22.         throw new MissingImplementationException("Replace with your specific ingress");
  23.     }
  24. }

When defined in yaml, routers are defined by a list of function types. The id component of the address is pulled from the key associated with each record in its underlying source implementation.

  1. targets:
  2.     - example-namespace/my-function-1
  3.     - example-namespace/my-function-2

Egress

Egress is the opposite of ingress; it is a point that takes messages and writes them to external systems. Each egress is defined using two components, an EgressIdentifier and an EgressSpec.

An egress identifier uniquely identifies an egress based on a namespace, name, and producing type. An egress spec defines the details of how to connect to the external system, the details are specific to each individual I/O module. Each identifier-spec pair are bound to the system inside a stateful function module.

  1. package org.apache.flink.statefun.docs.io.egress;
  3. import org.apache.flink.statefun.docs.models.User;
  4. import org.apache.flink.statefun.sdk.io.EgressIdentifier;
  6. public final class Identifiers {
  8.     public static final EgressIdentifier<User> EGRESS =
  9.         new EgressIdentifier<>("example", "egress", User.class);
  10. }
  1. package org.apache.flink.statefun.docs.io.egress;
  3. import java.util.Map;
  4. import org.apache.flink.statefun.docs.io.MissingImplementationException;
  5. import org.apache.flink.statefun.docs.models.User;
  6. import org.apache.flink.statefun.sdk.io.EgressIdentifier;
  7. import org.apache.flink.statefun.sdk.io.EgressSpec;
  8. import org.apache.flink.statefun.sdk.spi.StatefulFunctionModule;
  10. public class ModuleWithEgress implements StatefulFunctionModule {
  12.     @Override
  13.     public void configure(Map<String, String> globalConfiguration, Binder binder) {
  14.         EgressSpec<User> spec = createEgress(Identifiers.EGRESS);
  15.         binder.bindEgress(spec);
  16.     }
  18.     public EgressSpec<User> createEgress(EgressIdentifier<User> identifier) {
  19.         throw new MissingImplementationException("Replace with your specific egress");
  20.     }
  21. }
  1. version: "1.0"
  3. module:
  4.     meta:
  5.         type: remote
  6.     spec:
  7.         egresses:
  8.           - egress:
  9.               meta:
  10.                 id: example/user-egress
  11.                 type: # egress type
  12.               spec: # egress specific configurations

Stateful functions may then message an egress the same way they message another function, passing the egress identifier as function type.

  1. package org.apache.flink.statefun.docs.io.egress;
  3. import org.apache.flink.statefun.docs.models.User;
  4. import org.apache.flink.statefun.sdk.Context;
  5. import org.apache.flink.statefun.sdk.StatefulFunction;
  7. /** A simple function that outputs messages to an egress. */
  8. public class FnOutputting implements StatefulFunction {
  10.     @Override
  11.     public void invoke(Context context, Object input) {
  12.         context.send(Identifiers.EGRESS, new User());
  13.     }
  14. }