DataStream API 转换算子（Transform）

Map

        // 传入匿名类，实现MapFunction
        stream.map(new MapFunction<Event, String>() {
            @Override
            public String map(Event e) throws Exception {
                return e.user;
            }
        });
        
        // 传入Lambda表达式
        SingleOutputStreamOperator<String> result3 = stream.map(data -> data.user);

Filter

        // 传入匿名类实现FilterFunction
        stream.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event e) throws Exception {
                return e.user.equals("Mary");
            }
        });

        // 传入FilterFunction实现类
        stream.filter(new UserFilter()).print();
        
        // 传入Lambda表达式
        stream.filter(data->data.user.equals("Alice"));
        
    public static class UserFilter implements FilterFunction<Event> {
        @Override
        public boolean filter(Event e) throws Exception {
            return e.user.equals("Mary");
        }
    }

FlatMap

        stream.flatMap(new MyFlatMap()).print();
        
        stream.flatMap((Event value, Collector<String> out) -> {
            if (value.user.equals("Mary")) {
                out.collect(value.url);
            } else if (value.user.equals("Bob")) {
                out.collect(value.user);
                out.collect(value.url);
            }
        }).returns(new TypeHint<String>() {});
        
    public static class MyFlatMap implements FlatMapFunction<Event, String> {
        @Override
        public void flatMap(Event value, Collector<String> out) throws Exception {
            if (value.user.equals("Mary")) {
                out.collect(value.user);
            } else if (value.user.equals("Bob")) {
                out.collect(value.user);
                out.collect(value.url);
            }
        }
    }

KeyBy

KeyedStream<Tuple2<String, Integer>, String> wordAndOneKeyedStream = wordAndOneTuple.keyBy(data -> data.f0);

简单聚合

        // 安键分组之后进行聚合，提取当前用户最近一次访问数据
        stream.keyBy(new KeySelector<Event, String>() {

            @Override
            public String getKey(Event value) throws Exception {
                return value.user;
            }
        }).max("timestamp");
        
        // 安键分组之后进行聚合，提取当前用户最近一次访问数据
        stream.keyBy(data -> data.user).maxBy("timestamp");

max 其他字段用的是第一条数据，maxby是当前最大值对应其他字段值。

Reduce

对于一组数据，我们可以先取两个进行合并，然后再 将合并的结果看作一个数据、再跟后面的数据合并，最终会将它“简化”成唯一的一个数据， 这也就是 reduce“归约”的含义。在流处理的底层实现过程中，实际上是将中间“合并的结果” 作为任务的一个状态保存起来的;之后每来一个新的数据，就和之前的聚合状态进一步做归约。

计算最活跃用户：

                // 将Event数据类型转换成元组类型
                .map(new MapFunction<Event, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(Event e) throws Exception {
                        return Tuple2.of(e.user, 1L);
                    }
                })
                .keyBy(r -> r.f0) // 使用用户名来进行分流
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                        // 每到一条数据，用户pv的统计值加1
                        return Tuple2.of(value1.f0, value1.f1 + value2.f1);
                    }
                })
                .keyBy(r -> true) // 为每一条数据分配同一个key，将聚合结果发送到一条流中去
                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                        // 将累加器更新为当前最大的pv统计值，然后向下游发送累加器的值
                        return value1.f1 > value2.f1 ? value1 : value2;
                    }
                })

富函数类

        // 将点击事件转换成长整型的时间戳输出
        clicks.map(new RichMapFunction<Event, Long>() {
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        System.out.println("索引为 " + getRuntimeContext().getIndexOfThisSubtask() + " 的任务开始");
                    }

                    @Override
                    public Long map(Event value) throws Exception {
                        return value.timestamp;
                    }

                    @Override
                    public void close() throws Exception {
                        super.close();
                        System.out.println("索引为 " + getRuntimeContext().getIndexOfThisSubtask() + " 的任务结束");
                    }
                })

物理分区

        // 1. 随机分区
        stream.shuffle().print("shuffle").setParallelism(4);
        
        // 2. 轮询分区
        stream.rebalance().print("rebalance").setParallelism(4);
        
        // 3. rescale重缩放分区
        // 分成了几组，在自己的小组内轮询发送，可以认为是在某些场景下对rebalance的优化
        env.addSource(new RichParallelSourceFunction<Integer>() {  // 这里使用了并行数据源的富函数版本
                    @Override
                    public void run(SourceContext<Integer> sourceContext) throws Exception {
                        for (int i = 1; i <= 8; i++) {
                            // 将奇数发送到索引为1的并行子任务
                            // 将偶数发送到索引为0的并行子任务
                            if ( i % 2 == getRuntimeContext().getIndexOfThisSubtask()) {
                                sourceContext.collect(i);
                            }
                        }
                    }

                    @Override
                    public void cancel() {

                    }
                }) 
                .setParallelism(2)
                .rescale()
                .print().setParallelism(4);
                
        // 4. 广播，把当前的数据分发到下游所有的并行子任务
        stream.broadcast().print("broadcast").setParallelism(4);

        // 5. 全局分区，所有数据分配到一个分区
        stream.global().print("global").setParallelism(4);

        // 6. 自定义重分区
        // 将自然数按照奇偶分区
        env.fromElements(1, 2, 3, 4, 5, 6, 7, 8)
                .partitionCustom(new Partitioner<Integer>() {
                    @Override
                    public int partition(Integer key, int numPartitions) {
                        return key % 2;
                    }
                }, new KeySelector<Integer, Integer>() {
                    @Override
                    public Integer getKey(Integer value) throws Exception {
                        return value;
                    }
                })
                .print().setParallelism(2);