~~~ package com.gosuncn; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.sink.SinkFunction; public class WordCountStreamingJob { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource<String> dataStreamSource = env.fromElements("Apollo 1", "Paul 3", "Tom 2", "Paul 1", "Apollo 2", "Tom 4", "Marry 6", "Apollo 5", "Apollo 4", "Apollo 3"); SingleOutputStreamOperator<Tuple2> stream = dataStreamSource.map((MapFunction<String, Tuple2>) item -> Tuple2.of(item, 1)).returns(Types.TUPLE(Types.STRING, Types.INT)); stream.addSink(new CustomSinkFunction()); env.execute("WordCountStreamingJob"); } } /** * 自定義SinkFunction */ class CustomSinkFunction implements SinkFunction { @Override public void invoke(Object value, Context context) throws Exception { System.out.println(value); } } /** * 自定義RichSinkFunction */ class CustomSinkFunction extends RichSinkFunction { @Override public void invoke(Object value, Context context) throws Exception { // TODO 業務邏輯 } } ~~~