Apache Flink 学习笔记（二）

上一篇 Apache Flink 学习笔记（一） 简单示范了批处理的使用，本篇展示流式处理的使用方法。

流处理也叫无界处理，因为数据是源源不断的被加载进来的，流处理需要用到DataStream类。本篇demo 将结合kafka（公司有现成的消息生产者）来演示。

kafka 消息体如下（json）：

{
    "appId":"xxxx",
    "module":"xxxx"
    //其余省略
}

现在我想每10s统计一次，按照appid分组计数（需求简单一点），Event Time为ProcessingTime，Windows为滚动窗口。

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

public class Demo3 {
    public static void main(String[] args) {
        //生成流式执行环境对象 StreamExecutionEnvironment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().enableSysoutLogging();//开启Sysout打日志
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); //设置窗口的时间单位为process time
        env.setParallelism(2);//全局并发数
        //配置kafka bootstrap.servers
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "kafka bootstrap.servers");
        //配置消息主题和应用名（自定义工具类FlinkKafkaManager，源码在后面）
        FlinkKafkaManager manager = new FlinkKafkaManager("kafka.topic", "app.name", properties);
        //用JsonObject 反序列化接收kafka
        FlinkKafkaConsumer09<JSONObject> consumer = manager.build(JSONObject.class);
        //从最新的消息开始接收
        consumer.setStartFromLatest();
        //获得DataStream
        DataStream<JSONObject> messageStream = env.addSource(consumer);
        //转化为pojo
        DataStream<Bean3> bean3DataStream = messageStream.map(new FlatMap());
        bean3DataStream
                .keyBy(Bean3::getAppId) //也可以用“appId”替换
                .timeWindow(Time.seconds(10))//等价于下面这一行，因为上面设置了TimeCharacteristic.ProcessingTime
               // .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))//基于process time的窗口
                .aggregate(new Agg()) //聚合函数，这里也可以参照demo2用reduce函数
                .addSink(new Sink()); //输出函数
        try {
            env.execute("app.name");//流式处理需要调用触发
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class FlatMap implements MapFunction<JSONObject, Bean3> {
        @Override
        public Bean3 map(JSONObject jsonObject) throws Exception {
            return new Bean3(jsonObject.getString("appId"), jsonObject.getString("module"));
        }
    }

    public static class Agg implements AggregateFunction<Bean3, Tuple2<Bean3, Long>, Tuple2<Bean3, Long>> {
        @Override
        public Tuple2<Bean3, Long> createAccumulator() {
            return new Tuple2<Bean3, Long>();
        }

        @Override
        public Tuple2<Bean3, Long> add(Bean3 bean3, Tuple2<Bean3, Long> bean3LongTuple2) {
            Bean3 bean = bean3LongTuple2.f0;
            Long count = bean3LongTuple2.f1;
            if (bean == null) {
                bean = bean3;
            }
            if (count == null) {
                count = 1L;
            } else {
                count++;
            }
            return new Tuple2<>(bean, count);
        }

        @Override
        public Tuple2<Bean3, Long> getResult(Tuple2<Bean3, Long> bean3LongTuple2) {
            return bean3LongTuple2;
        }

        @Override
        public Tuple2<Bean3, Long> merge(Tuple2<Bean3, Long> bean3LongTuple2, Tuple2<Bean3, Long> acc1) {
            Bean3 bean = bean3LongTuple2.f0;
            Long count = bean3LongTuple2.f1;
            Long acc = acc1.f1;
            return new Tuple2<>(bean, count + acc);
        }
    }

    public static class Sink implements SinkFunction<Tuple2<Bean3, Long>> {
        @Override
        public void invoke(Tuple2<Bean3, Long> value, Context context) throws Exception {
            System.out.println(value.f0.toString() + "," + value.f1);
        }
    }

    public static class Bean3 {
        public String appId;
        public String module;

        public Bean3() {
        }

        public Bean3(String appId, String module) {
            this.appId = appId;
            this.module = module;
        }

        public String getAppId() {
            return appId;
        }

        public void setAppId(String appId) {
            this.appId = appId;
        }

        public String getModule() {
            return module;
        }

        public void setModule(String module) {
            this.module = module;
        }

        @Override
        public String toString() {
            return "Bean3{" +
                    "appId='" + appId + '\'' +
                    ", module='" + module + '\'' +
                    '}';
        }
    }
}

与上一篇批处理的demo相比，流处理显得复杂了许多。实际上二者有很多想通的地方，比如批处理中的groupBy和流处理的keyBy，都是按照指定维度分组的。

而流处理中会引入窗口的概念，正如前面所说，流式数据是无界数据，Flink 借助窗口将无界数据转化成一个个“批处理”再做计算。窗口分为滚动窗口，滑动窗口，会话窗口等等，具体可参见官网介绍。而每个窗口的时间划分则是由event time 决定的，本例采用的是ProcessingTime即处理时间。

下面我将demo3改造，使其变成使用EventTime，也就是说窗口的时间由数据源的时间戳（事件发生）决定。

改动1

//为pojo Bean3 添加时间戳字段
public static class Bean3 {
    public Long timestamp;//add event time
    public String appId;
    public String module;

    public Bean3() {
    }

    public Bean3(Long timestamp, String appId, String module) {
        this.timestamp = timestamp;
        this.appId = appId;
        this.module = module;
    }

    public long getTimestamp() {
        return timestamp;
    }

    public void setTimestamp(Long timestamp) {
        this.timestamp = timestamp;
    }
//省略其他
}

改动2

//设置窗口的时间单位为event time  
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 

改动3

//新增
//指定数据源的时间戳，Time.seconds(int)是指允许多长时间消息延迟
DataStream<Bean3> bean3DataStreamWithAssignTime = 
bean3DataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Bean3>(Time.seconds(0)) {
    @Override
    public long extractTimestamp(Bean3 element) {
        return element.getTimestamp();
    }
});

改动4

bean3DataStreamWithAssignTime
                .keyBy(Bean3::getAppId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))//基于event time的窗口
                .allowedLateness(Time.seconds(5)) //允许数据延迟多长时间,谨慎使用,迟到的数据会导致出现重复
//后面省略

FlinkKafkaManager 源码

package flink.test.manager;

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import java.util.Properties;

public class FlinkKafkaManager<T> {
    private String topic;
    private String groupId;
    private Properties properties;

    public FlinkKafkaManager(String topic, String groupId, Properties properties) {
        this.topic = topic;
        this.groupId = groupId;
        this.properties = properties;
        this.properties.setProperty("group.id", this.groupId);
        //为使用默认kafka的用户配置基础配置
        this.setDefaultKafkaProperties();
    }

    private void setDefaultKafkaProperties() {
        //启用auto commit offset, 每5s commit一次
        this.properties.setProperty("enable.auto.commit", "true");
        this.properties.setProperty("auto.commit.interval.ms", "5000");
    }

    public FlinkKafkaConsumer09<T> build(Class<T> clazz) {
        if (checkProperties()) {
            return new FlinkKafkaConsumer09<T>(topic, new ConsumerDeserializationSchema(clazz), properties);
        } else {
            return null;
        }
    }

    private boolean checkProperties() {
        boolean isValued = true;

        if (!properties.containsKey("bootstrap.servers")) {
            isValued = false;
        } else {
            String brokers = properties.getProperty("bootstrap.servers");
            if (brokers == null || brokers.isEmpty()) {
                isValued = false;
            }
        }

        if (this.topic == null || this.topic.isEmpty()) {
            isValued = false;
        }

        if (!properties.containsKey("group.id")) {
            isValued = false;
        } else {
            String groupId = properties.getProperty("group.id");
            if (groupId == null || groupId.isEmpty()) {
                isValued = false;
            }
        }

        return isValued;
    }
}

ConsumerDeserializationSchema 源码

package flink.test.manager;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;

import java.io.IOException;

public class ConsumerDeserializationSchema<T> implements DeserializationSchema<T> {
    private Class<T> clazz;

    public ConsumerDeserializationSchema(Class<T> clazz) {
        this.clazz = clazz;
    }

    @Override
    public T deserialize(byte[] bytes) throws IOException {
        //确保 new String(bytes) 是json 格式，如果不是，请自行解析
        return JSON.parseObject(new String(bytes), clazz);
    }

    @Override
    public boolean isEndOfStream(T t) {
        return false;
    }

    @Override
    public TypeInformation<T> getProducedType() {
        return TypeExtractor.getForClass(clazz);
    }
}