Flink在实时在实时计算平台和实时数仓中的企业级应用小结

我是抖音主播，我想看带货销售情况的排行？我是运营，我想看到我们公司销售商品的 TOP10？我是开发，我想看到我们公司所有生产环境中服务器的运行情况？...... 

tar zxf apache-flume-1.8.0-bin.tar.gz

# 定义这个 agent 中各组件的名字a1.sources = r1a1.sinks = k1a1.channels = c1
# source的配置，监听日志文件中的新增数据a1.sources.r1.type = execa1.sources.r1.command  = tail -F /home/logs/access.log
#sink配置，使用avro日志做数据的消费a1.sinks.k1.type = avroa1.sinks.k1.hostname = flumeagent03a1.sinks.k1.port = 9000
#channel配置，使用文件做数据的临时缓存a1.channels.c1.type = filea1.channels.c1.checkpointDir = /home/temp/flume/checkpointa1.channels.c1.dataDirs = /home/temp/flume/data
#描述和配置 source channel sink 之间的连接关系a1.sources.r1.channels = c1a1.sinks.k1.channel = c

$ flume-ng agent -c conf -n a1 -f conf/log_kafka.conf >/dev/null 2>&1 &

# 定义这个 agent 中各组件的名字a1.sources = r1a1.sinks = k1a1.channels = c1
#source配置a1.sources.r1.type = avroa1.sources.r1.bind = 0.0.0.0a1.sources.r1.port = 9000
#sink配置a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSinka1.sinks.k1.topic = log_kafkaa1.sinks.k1.brokerList = 127.0.0.1:9092a1.sinks.k1.requiredAcks = 1a1.sinks.k1.batchSize = 20
#channel配置a1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100
#描述和配置 source channel sink 之间的连接关系a1.sources.r1.channels = c1a1.sinks.k1.channel = c1    

$ flume-ng agent -c conf -n a1 -f conf/flume_kafka.conf >/dev/null 2>&1 &

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);// 检查点配置,如果要用到状态后端，那么必须配置env.setStateBackend(new MemoryStateBackend(true));
Properties properties = new Properties();properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
properties.setProperty(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, "10");FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("log_user_action", new SimpleStringSchema(), properties);//设置从最早的offset消费consumer.setStartFromEarliest();
DataStream<UserClick> dataStream = env        .addSource(consumer)        .name("log_user_action")        .map(message -> {            JSONObject record = JSON.parseObject(message);            return new UserClick(                    record.getString("user_id"),                    record.getLong("timestamp"),                    record.getString("action")            );        })        .returns(TypeInformation.of(UserClick.class));

SingleOutputStreamOperator<UserClick> userClickSingleOutputStreamOperator = dataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<UserClick>(Time.seconds(30)) {    @Override    public long extractTimestamp(UserClick element) {        return element.getTimestamp();    }});

dataStream     .windowAll(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8))).trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(20)))

userClickSingleOutputStreamOperator         .keyBy(new KeySelector<UserClick, String>() {            @Override            public String getKey(UserClick value) throws Exception {                return DateUtil.timeStampToDate(value.getTimestamp());            }        })        .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)))        .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(20)))        .evictor(TimeEvictor.of(Time.seconds(0), true))        ...

public class MyProcessWindowFunction extends ProcessWindowFunction<UserClick,Tuple3<String,String, Integer>,String,TimeWindow>{
    private transient MapState<String, String> uvState;    private transient ValueState<Integer> pvState;
    @Override    public void open(Configuration parameters) throws Exception {
        super.open(parameters);        uvState = this.getRuntimeContext().getMapState(new MapStateDescriptor<>("uv", String.class, String.class));        pvState = this.getRuntimeContext().getState(new ValueStateDescriptor<Integer>("pv", Integer.class));    }
    @Override    public void process(String s, Context context, Iterable<UserClick> elements, Collector<Tuple3<String, String, Integer>> out) throws Exception {
        Integer pv = 0;        Iterator<UserClick> iterator = elements.iterator();        while (iterator.hasNext()){            pv = pv + 1;            String userId = iterator.next().getUserId();            uvState.put(userId,null);        }        pvState.update(pvState.value() + pv);
        Integer uv = 0;        Iterator<String> uvIterator = uvState.keys().iterator();        while (uvIterator.hasNext()){            String next = uvIterator.next();            uv = uv + 1;        }
        Integer value = pvState.value();        if(null == value){            pvState.update(pv);        }else {            pvState.update(value + pv);        }
        out.collect(Tuple3.of(s,"uv",uv));        out.collect(Tuple3.of(s,"pv",pvState.value()));    }}

userClickSingleOutputStreamOperator        .keyBy(new KeySelector<UserClick, String>() {            @Override            public String getKey(UserClick value) throws Exception {                return value.getUserId();            }        })        .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)))        .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(20)))        .evictor(TimeEvictor.of(Time.seconds(0), true))        .process(new MyProcessWindowFunction());

...userClickSingleOutputStreamOperator            .keyBy(new KeySelector<UserClick, String>() {                @Override                public String getKey(UserClick value) throws Exception {                    return value.getUserId();                }            })            .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)))            .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(20)))            .evictor(TimeEvictor.of(Time.seconds(0), true))            .process(new MyProcessWindowFunction())            .addSink(new RedisSink<>(conf,new MyRedisSink()));...