I am following https://github.com/immerok/recipes/ examples and found them very useful! I am trying to figure out what’s the best approach to pass the DataStream into a processor or a function for further data analysis. Does sinkTo method sounds like a possible solution? For example, in TableDeduplicatedJoin.java:

defineWorkflow(

tableEnv,

customerStream,

transactionStream,

workflow -> workflow.sinkTo(new PrintSink<>()));

sinkTo, passes dataStream for print (PrintSink). Should I create a new class (i.e. DataProcessingSink implements Sink) and override

createWriter

method?

Checkout the docs for streaming operators: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/datastream/operators/overview/ Sinks are terminal operators used to send data to external systems. You might be better off with a process/map function

How about ProcessFunction? Something like this:

public class EMPADProcessor extends ProcessFunction<Row, Row> {

    @Override
    public void processElement(Row row, ProcessFunction<Row, Row>.Context context,
                               Collector<Row> collector) throws Exception {

        // process row here...

    }
}

where:

tableEnv.toChangelogStream(resultTable).process(new EMPADProcessor());

Yep that works