Flink 对算子的Function接口实现

1.创建类实现Function接口

通过新建类去实现MapFunction，并自定义数据处理逻辑。

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
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.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import java.time.Duration;
import org.apache.flink.streaming.api.datastream.DataStream;
//用类去实现function
class MyMapFunction implements MapFunction
{
    @Override
    public String map(String value) throws Exception
    {
        return value.toUpperCase();
    }
};
public class test {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.创建Class实现Function接口
        DataStreamSource inputStream = env.socketTextStream("localhost", 7777);
        DataStream map_result=inputStream.map(new MyMapFunction());
        map_result.print();
        env.execute("run map");

    }
}
;

2.利用匿名类去实现function

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
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.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import java.time.Duration;
import org.apache.flink.streaming.api.datastream.DataStream;
//创建匿名类去实现function
public class test2{
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.创建Class实现Function接口
        DataStreamSource inputStream = env.socketTextStream("localhost", 7777);
        DataStream map_result=inputStream.map(
                new MapFunction()
                {
            @Override
            public String map(String value) throws Exception
            {
                return value.toUpperCase();
            }
                }
        );
        map_result.print();
        env.execute("run map");

    }
}
;