Flink算子(Map,FlatMap,Filter) · flink

# Transformation ## Map DataStreamSource<T> 轉換 SingleOutputStreamOperator<T> ```java public class WordCountStreamingJob { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource<Integer> streamSource = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9)); SingleOutputStreamOperator<Integer> result = streamSource.map(new MapFunction<Integer, Integer>() { @Override public Integer map(Integer item) throws Exception { return item * 1; } }); result.print(); env.execute("WordCountStreamingJob"); } } ``` 比如，傳入的是一個整型的集合，傳出的是另一個整型的集合。一比一對應。簡化版： ```java public class WordCountStreamingJob { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource<Integer> streamSource = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9)); SingleOutputStreamOperator<Integer> result = streamSource.map((MapFunction<Integer, Integer>) item -> item * 1); result.print(); env.execute("WordCountStreamingJob"); } } ``` ## FlatMap ```java public class WordCountStreamingJob { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource<Integer> streamSource = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9)); streamSource.flatMap(new FlatMapFunction<Integer, Integer>() { @Override public void flatMap(Integer item, Collector<Integer> out) throws Exception { // 復制十份 for (int i = 0; i < 10; i++) { out.collect(item); } } }).print(); env.execute("WordCountStreamingJob"); } } ``` 傳入一個集合，把每個集合復制十份，再輸出一個集合。 ## Filter ```java public class WordCountStreamingJob { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource<Integer> streamSource = env.fromCollection(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9)); streamSource.filter((FilterFunction<Integer>) item -> item > 5).print(); env.execute("WordCountStreamingJob"); } } ``` 返回一個布爾值，來過濾數據。