我是抖音主播，我想看帶貨銷售情況的排行？我是運(yùn)營(yíng)，我想看到我們公司銷售商品的 TOP10？我是開(kāi)發(fā)，我想看到我們公司所有生產(chǎn)環(huán)境中服務(wù)器的運(yùn)行情況？...... 

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

# 定義這個(gè) agent 中各組件的名字a1.sources = r1a1.sinks = k1a1.channels = c1
# source的配置，監(jiān)聽(tīng)日志文件中的新增數(shù)據(jù)a1.sources.r1.type = execa1.sources.r1.command  = tail -F /home/logs/access.log
#sink配置，使用avro日志做數(shù)據(jù)的消費(fèi)a1.sinks.k1.type = avroa1.sinks.k1.hostname = flumeagent03a1.sinks.k1.port = 9000
#channel配置，使用文件做數(shù)據(jù)的臨時(shí)緩存a1.channels.c1.type = filea1.channels.c1.checkpointDir = /home/temp/flume/checkpointa1.channels.c1.dataDirs = /home/temp/flume/data
#描述和配置 source channel sink 之間的連接關(guān)系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 &

# 定義這個(gè) 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 之間的連接關(guān)系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);// 檢查點(diǎn)配置,如果要用到狀態(tài)后端，那么必須配置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);//設(shè)置從最早的offset消費(fèi)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()));...