Flink

@Bean
public DataSetCallback<Long> dataSetCallback() {
    return new DataSetCallback<Long>() {
        public Long onDataSet(DataSet dataSet, Object... objects) {
            try {
                 dataSet.print();
                 return new Long(0);
            } catch (Exception e) {
                 return new Long(-1);
            }
        }
    };
}

@Bean
public VoidDataStreamCallback dataStreamCallback() {
    return new VoidDataStreamCallback() {
        @Override
        public void doOnDataStream(DataStream dataStream, Object... objects) throws Exception {
            dataStream.flatMap(new Splitter()).print();

            environment.execute("data stream test");
        }
    };
}

CamelContext camelContext = new SpringCamelContext(context);

String pattern = "foo";

try {
    ProducerTemplate template = camelContext.createProducerTemplate();
    camelContext.start();
    Long count = template.requestBody("flink:dataSet?dataSet=#myDataSet&dataSetCallback=#countLinesContaining", pattern, Long.class);
    } finally {
        camelContext.stop();
    }

@Bean
public StreamExecutionEnvironment streamExecutionEnvironment() {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // Configure for batch processing
    env.setRuntimeMode(RuntimeExecutionMode.BATCH);
    return env;
}

@Bean
public DataStreamSource<String> myDataStream(StreamExecutionEnvironment env) {
    return env.readTextFile("src/test/resources/testds.txt");
}

@Bean
public DataStreamCallback wordCountCallback() {
    return new VoidDataStreamCallback() {
        @Override
        public void doOnDataStream(DataStream dataStream, Object... payloads) throws Exception {
            dataStream
                .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {
                    for (String word : line.split("\\s+")) {
                        out.collect(Tuple2.of(word, 1));
                    }
                })
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(tuple -> tuple.f0)
                .sum(1)
                .print();
        }
    };
}

// In your route
from("direct:wordCount")
    .to("flink:datastream?dataStream=#myDataStream&dataStreamCallback=#wordCountCallback");

@Bean
public StreamExecutionEnvironment streamingEnvironment() {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // Configure for streaming (default mode)
    env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
    // Enable checkpointing for fault tolerance
    env.enableCheckpointing(10000); // checkpoint every 10 seconds
    return env;
}

@Bean
public DataStreamCallback streamingProcessCallback() {
    return new VoidDataStreamCallback() {
        @Override
        public void doOnDataStream(DataStream dataStream, Object... payloads) throws Exception {
            dataStream
                .map(event -> processEvent(event))
                .keyBy(event -> event.getKey())
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
                .aggregate(new MyAggregateFunction())
                .addSink(new MyCustomSink());

            // Execute the streaming job
            dataStream.getExecutionEnvironment().execute("Streaming Job");
        }
    };
}