HotNItems

　　拓展需求：实时统计双十一下单量，实时统计成交额，实时查看锅炉温度变化曲线，每个5分钟看一下过去一个小时温度变化曲线，

　　涉及到的技术点：sliding window、Watermark、event time

　　用到的算子或者说叫链式调用：keyby、timeWindow、aggregate、assignTimestampsAndWatermarks、filter、processFunction底层API

PopularPlacesToEs

　　框架：flume -> Kafka、flink、es、kibana

　　涉及到的技术点：sliding window、watermark、event time

　　用到的算子：keyby、filter、apply、map、timeWindow

实现一个“实时热门商品”的需求，我们可以将“实时热门商品”翻译成程序员更好理解的需求：

　　每隔5分钟输出最近一小时内点击量最多的前N个商品。将这个需求进行分解我们大概要做这么几件事情：

• 抽取出业务时间戳，告诉Flink框架基于业务时间做窗口

• 过滤出点击行为数据

• 按一小时的窗口大小，每5分钟统计一次，做滑动窗口聚合（Sliding Window）

• 按每个窗口聚合，输出每个窗口中点击量前N名的商品

public class HotItems {
public static void main(String[] args) throws Exception {
//创建执行环境 execution environment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 告诉系统按照 EventTime 处理
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
// 为了打印到控制台的结果不乱序，我们配置全局的并发为1，改变并发对结果正确性没有影响
//env.setMaxParallelism(1);
//Caused by: org.apache.flink.runtime.JobException: Vertex Split Reader: Custom File source -> Timestamps/Watermarks -> Filter's parallelism (8) is higher than the max parallelism (1). Please lower the parallelism or increase the max parallelism.
env.setParallelism(1);
// UserBehavior.csv 的本地文件路径, 在 resources 目录下
URL fileURL = HotItems.class.getClassLoader().getResource("UserBehavior.csv");
Path filePath = Path.fromLocalFile(new File(fileURL.toURI())); //抛出异常URISyntaxException

    // 抽取 UserBehavior 的 TypeInformation，是一个 PojoTypeInfo ???  
    PojoTypeInfo<UserBehavior> pojoType = (PojoTypeInfo<UserBehavior>) TypeExtractor.createTypeInfo(UserBehavior.class);//TypeInformation<UserBehavior>  
    String\[\] fieldOrder = new String\[\]{"userId", "itemId", "categoryId", "behavior", "timestamp"};

    //创建PojoCsvInputFormat  
    PojoCsvInputFormat<UserBehavior> csvInput = new PojoCsvInputFormat<>(filePath, pojoType, fieldOrder);  
    // 创建数据源，得到 UserBehavior 类型的 DataStream  
    env.createInput(csvInput, pojoType)  
            .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {  
                @Override  
                public long extractAscendingTimestamp(UserBehavior userBehavior) {  
                    return userBehavior.timestamp \* 1000;// 原始数据单位秒，将其转成毫秒  
                }  
            }).filter(new FilterFunction<UserBehavior>() { // 过滤出只有点击的数据  
        @Override  
        public boolean filter(UserBehavior userBehavior) throws Exception {  
            return userBehavior.behavior.equals("pv");  
        }  
    }).keyBy("itemId")// 我们使用.keyBy("itemId")对商品进行分组聚合  
            // 使用.timeWindow(Time size, Time slide)对每个商品做滑动窗口（1小时窗口，5分钟滑动一次）。  
            .timeWindow(Time.minutes(60), Time.minutes(5)) //别导错包了  
            .aggregate(new CountAgg(), new WindowResultFunction())  
            //CountAgg统计窗口中的条数； 商品ID，窗口，点击量封装成了ItemViewCount进行输出  
            .keyBy("windowEnd")  
            .process(new TopNHotItems(3)).print();

    env.execute("Hot Items job");

}  
/\*\*  
 \* 求某个窗口中前 N 名的热门点击商品，key 为窗口时间戳，输出为 TopN 的结果字符串  
 \*/  
public static class TopNHotItems extends KeyedProcessFunction<Tuple, ItemViewCount, String> {  
    private final int topSize;  
    public TopNHotItems(int topSize) throws Exception {  
        this.topSize = topSize;  
    }

    // 用于存储商品与点击数的状态，待收齐同一个窗口的数据后，再触发 TopN 计算  
    private ListState<ItemViewCount> itemState;

    /\*  
     \* 这里我们还使用了ListState<ItemViewCount>来存储收到的每条ItemViewCount消息，  
     \* 保证在发生故障时，状态数据的不丢失和一致性。  
     \* ListState是Flink提供的类似Java List接口的State API，  
     \* 它集成了框架的checkpoint机制，自动做到了exactly-once的语义保证。\*/  
    @Override  
    public void open(Configuration parameters) throws Exception {  
        super.open(parameters);  
        ListStateDescriptor<ItemViewCount> itemsStateDesc = new ListStateDescriptor<>(  
                "itemState-state",//状态的描述符  
                ItemViewCount.class);//存储的类型  
        //从运行时上下文获取  
        itemState = getRuntimeContext().getListState(itemsStateDesc);  
    }  
    /\*         \* ProcessFunction是Flink提供的一个low-level API，用于实现更高级的功能。  
     \* 它主要提供了定时器timer的功能（支持EventTime或ProcessingTime）。  
     \* 本案例中我们将利用timer来判断何时收齐了某个window下所有商品的点击量数据。  
     \* 由于Watermark的进度是全局的，在processElement方法中，每当收到一条数据（ItemViewCount），我们就注册一个windowEnd+1的定时器（Flink框架会自动忽略同一时间的重复注册）。  
     \* windowEnd+1的定时器被触发时，意味着收到了windowEnd+1的Watermark，即收齐了该windowEnd下的所有商品窗口统计值。  
     \* 我们在onTimer()中处理将收集的所有商品及点击量进行排序，选出TopN，并将排名信息格式化成字符串后进行输出。\*/

    @Override  
    public void processElement(ItemViewCount input, Context context, Collector<String> collector) throws Exception {  
        // 每条数据都保存到状态中  
        itemState.add(input);  
        // 注册 windowEnd+1 的 EventTime Timer, 当触发时，说明收齐了属于windowEnd窗口的所有商品数据  
        context.timerService().registerEventTimeTimer(input.windowEnd + 1);  
    }

    @Override  
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {  
        // 获取收到的所有商品点击量  
        List<ItemViewCount> allItems = new ArrayList<>();  
        for (ItemViewCount item : itemState.get()) {  
            allItems.add(item);  
        }  
        // 提前清除状态中的数据，释放空间  
        itemState.clear();  
        // 按照点击量从大到小排序  
        allItems.sort(new Comparator<ItemViewCount>() {  
            @Override  
            public int compare(ItemViewCount o1, ItemViewCount o2) {  
                return (int) (o2.viewCount - o1.viewCount);  
            }  
        });  
        // 将排名信息格式化成 String, 便于打印  
        StringBuilder result = new StringBuilder();  
        result.append("===========================\\n");  
        result.append("时间：").append(new Timestamp(timestamp - 1)).append("\\n");  
        for (int i = 0; i < allItems.size() && i < topSize; i++) {  
            ItemViewCount currentItem = allItems.get(i);  
            // No1:  商品ID=12224  浏览量=2413  
            result.append("No").append(i).append(":")  
                    .append(" 商品ID=").append(currentItem.itemId)  
                    .append(" 浏览量=").append(currentItem.viewCount)  
                    .append("\\n");  
        }  
        result.append("==========================\\n\\n");  
        // 控制输出频率，模拟实时滚动结果  
        Thread.sleep(1000);  
        out.collect(result.toString());  
        //super.onTimer(timestamp, ctx, out);  
    }  
}

/\*\* 用于输出窗口的结果 \*/  
/\*\* 将每个key每个窗口聚合后的结果带上其他信息进行输出。\*/  
/\*\*  
 \* 我们这里实现的WindowResultFunction将主键商品ID，窗口，点击量封装成了ItemViewCount进行输出。  
 \*/

private static class WindowResultFunction implements WindowFunction<Long, ItemViewCount, Tuple, TimeWindow> {  
    @Override  
    public void apply(Tuple key,  // 窗口的主键，即 itemId  
                      TimeWindow window, // 窗口  
                      Iterable<Long> aggregateResult,// 聚合函数的结果，即 count 值  
                      Collector<ItemViewCount> collector) // 输出类型为 ItemViewCount  
    {  
        Long itemId = ((Tuple1<Long>) key).f0;  
        Long count = aggregateResult.iterator().next();  
        collector.collect(ItemViewCount.of(itemId, window.getEnd(), count));

    }

}

/\*\*  
 \* 商品点击量(窗口操作的输出类型)  
 \*/  
public static class ItemViewCount {  //public  
    public long itemId;     // 商品ID  
    public long windowEnd;  // 窗口结束时间戳  
    public long viewCount;  // 商品的点击量

    public static ItemViewCount of(long itemId, long windowEnd, long viewCount) {  
        ItemViewCount result = new ItemViewCount();  
        result.itemId = itemId;  
        result.windowEnd = windowEnd;  
        result.viewCount = viewCount;  
        return result;  
    }  
}

/\*\* COUNT 统计的聚合函数实现，每出现一条记录加一 \*/  
/\*\* 接口: AggregateFunction(in, acc, out) \*/  
/\*\*  
 \* 这里的CountAgg实现了AggregateFunction接口，功能是统计窗口中的条数，即遇到一条数据就加一。  
 \*/  
public static class CountAgg implements AggregateFunction<UserBehavior, Long, Long> {

    @Override  
    public Long createAccumulator() {  
        return 0L;  
    }

    @Override  
    public Long add(UserBehavior userBehavior, Long acc) {  
        return acc + 1;  
    }

    @Override  
    public Long getResult(Long acc) {  
        return acc;  
    }

    @Override  
    public Long merge(Long acc1, Long acc2) {  
        return acc1 + acc2;  
    }  
}

/\*\*  
 \* 用户行为数据结构  
 \*\*/  
public static class UserBehavior {  
    public long userId;         // 用户ID  
    public long itemId;         // 商品ID  
    public int categoryId;      // 商品类目ID  
    public String behavior;     // 用户行为, 包括("pv", "buy", "cart", "fav")  
    public long timestamp;      // 行为发生的时间戳，单位秒

}  

}

===========================
时间：2017-11-26 09:05:00.0
No0: 商品ID=5051027 浏览量=3
No1: 商品ID=3493253 浏览量=3

No2: 商品ID=4261030 浏览量=3

===========================
时间：2017-11-26 09:10:00.0
No0: 商品ID=812879 浏览量=5
No1: 商品ID=2600165 浏览量=4

No2: 商品ID=2828948 浏览量=4

===========================
时间：2017-11-26 09:15:00.0
No0: 商品ID=812879 浏览量=7
No1: 商品ID=138964 浏览量=5

No2: 商品ID=4568476 浏览量=5