The FlinkStreamletLogic provides methods to read and write data streams, and provides access to the configuration of the streamlet, among other things. A FlinkStreamletLogic must provide a method for executing streaming queries that process the Flink computation graph. The method buildExecutionGraph has to be overridden by implementation classes that process one or more DataStream {empty}s. The resulting graph is then submitted by executeStreamingQueries to the Flink runtime StreamExecutionEnvironment to generate the final output. These jobs will be run by the run method of FlinkStreamlet to produce a StreamletExecution class, the StreamletExecution class manages the execution of a Flink Job. The FlinkStreamletLogic may contain instance values since it's only constructed in runtime. The Streamlet, however, is also instantiated during compile-time, to extract metadata, and must not contain any instance values.

Overide the method buildExecutionGraph to build the computation graph that needs to run as part of the business logic for the FlinkStreamlet.

Here's an example of how to provide a specialized implementation of FlinkStreamletLogic as part of implementing a custom FlinkStreamlet:

  // new custom `FlinkStreamlet`
  // define inlets, outlets and shape

  // provide custom implementation of `FlinkStreamletLogic`
  override def createLogic() = new FlinkStreamletLogic {
    override def buildExecutionGraph = {
      val ins: DataStream[Data] = readStream(in)
      val simples: DataStream[Simple] = ins.map(r => new Simple(r.getName()))
      writeStream(out, simples)
    }
  }
}