env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
此处可以取以下三类值：
-
Event Time
事件时间，事件(Event)本身的时间，即数据流中事件实际发生的时间，通常使用事件发生时的时间戳来描述，这些事件的时间戳通常在进入流处理应用之前就已经存在了，事件时间反映了事件真实的发生时间。所以，基于事件时间的计算操作，其结果是具有确定性的，无论数据流的处理速度如何、事件到达算子的顺序是否会乱，最终生成的结果都是一样的。
-
Ingestion Time
摄入时间，事件进入Flink的时间，即将每一个事件在数据源算子的处理时间作为事件时间的时间戳，并自动生成水位线(watermarks,关于watermarks下文会详细分析)。
Ingestion Time从概念上讲介于Event Time和Processing Time之间。与Processing Time相比 ，它的性能消耗更多一些，但结果却更可预测。由于 Ingestion Time使用稳定的时间戳（在数据源处分配了一次），因此对记录的不同窗口操作将引用相同的时间戳，而在Processing Time中每个窗口算子都可以将记录分配给不同的窗口。
与Event Time相比，Ingestion Time无法处理任何乱序事件或迟到的数据，即无法提供确定的结果，但是程序不必指定如何生成水位线。在内部，Ingestion Time与Event Time非常相似，但是可以实现自动分配时间戳和自动生成水位线的功能。
-
Processing Time
处理时间，根据处理机器的系统时钟决定数据流当前的时间，即事件被处理时当前系统的时间。还以窗口算子为例(关于window，下文会详细分析)，基于处理时间的窗口操作是以机器时间来进行触发的，由于数据到达窗口的速率不同，所以窗口算子中使用处理时间会导致不确定的结果。在使用处理时间时，无需等待水位线的到来后进行触发窗口，所以可以提供较低的延迟。
举一个实际的例子：
站在看电影的普通大众来说肯定是关心电影的上映时间；也就是处理时间 也就是processTime
站在研究电影剧情的人来说肯定关心的是电影整个系列的故事发展，应该先看1999的星球大战1....也就是eventTime。
另外再看一个例子：
eventTime 是追求数据准，接受一定的延迟
processTime是追求快，容忍数据存在一定的偏差
flink默认使用eventTime
实例package demo;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
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.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Properties;
public class HotItems {
public static void main(String[] args) throws Exception {
//创建环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
//wx 2021年10月11日16:48:33 EventTime是以数据自带的时间戳字段为准，应用程序需要指定如何从record中抽取时间戳字段，此处对应的是csv中的最后一个字段 - 行为发生的时间戳
//其他两个取值是:
// 1. ProcessingTime是以operator处理的时间为准，它使用的是机器的系统时间来作为data stream的时间；
// 2. IngestionTime是以数据进入flink streaming data flow的时间为准；
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
Properties ps = new Properties();
ps.setProperty("bootstrap.servers","192.168.149.131:9092");//集群地址
ps.setProperty("group.id", "consumer_group");//消费者组
ps.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//key反序列化方式
ps.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//value反序列化方式
ps.setProperty("auto.offset.reset","earliest");//消费策略
//其实第二个参数指定了序列化方式，那key和value的序列化方式就不用指定了
DataStream inputStream=env.addSource(new FlinkKafkaConsumer("hotitems_test",new SimpleStringSchema(),ps));
DataStream dataStream=inputStream.map(
line ->{
String [] words=line.split(",");
return new UserBehavior(new Long(words[0]),new Long(words[1]),new Integer(words[2]),new String(words[3]),new Long(words[4]));
})
.assignTimestampsAndWatermarks(
new AscendingTimestampExtractor() { //升序
@Override
public long extractAscendingTimestamp(UserBehavior userBehavior) {//获取事件时间戳，秒级转毫秒级
return userBehavior.getTimestamp()*1000L;
}
});
//分组聚合得到结果数据
DataStream aggStream=dataStream
.filter(data -> "pv".equals(data.getBehavior())) //过滤“pv”行为
.keyBy(UserBehavior::getItemId)
.timeWindow(Time.minutes(60),Time.minutes(5)) //每5分钟更新一次1小时窗口数据
.aggregate(new ItemCountAgg(),new WindowItemCountResult());
//收集同一窗口所有商品的count数据，按top 5输出
DataStream resultDs=aggStream
.keyBy("windowEnd")
.process(new TopNItems(5));
//输出并执行
resultDs.print("每隔五分钟最近一小时前五的热门商品");
env.execute("hot items analysis");
}
//泛型1:输入类型 泛型2:聚合状态类型 泛型3:输出类型
public static class ItemCountAgg implements AggregateFunction{
@Override
public Long createAccumulator() {//创建累加器并给初始值
return 0L;
}
@Override
public Long add(UserBehavior userBehavior,Long accumulator) {//每次计算累加器加一,并返回新的累加器值
return accumulator+1;
}
@Override
public Long getResult(Long accumulator) {//累加器最终给外部返回的值
return accumulator;
}
@Override
public Long merge(Long a, Long b) { //合并两个累加器，返回合并后的累加器的状态，这儿用不到.用不到.
return a+b;
}
}
//参数1：输入类型 参数2：输出类型 参数3：keyBy的返回值键值对中value的类型 参数4： 窗口类型
public static class WindowItemCountResult implements WindowFunction{
@Override
public void apply(Long key, TimeWindow window, Iterable iterable, Collector collector) throws Exception {
//包装成一个ItemViewCount对象输出
collector.collect(new ItemViewCount(key.longValue(),window.getEnd(),iterable.iterator().next()));
}
}
//参数1：keyBy返回值类型 参数2：输入类型 参数3：输出类型
public static class TopNItems extends KeyedProcessFunction{
private Integer topSize;
private ListState listState; //列表状态，保存当前窗口所有输出的ItemViewCount
public TopNItems(Integer topSize) {
this.topSize = topSize;
}
@Override
public void open(Configuration parameters) throws Exception {
listState =getRuntimeContext().getListState(new ListStateDescriptor("item-view-count-list",ItemViewCount.class));
}
//每来一条数据如何处理
@Override
public void processElement(ItemViewCount value, Context context, Collector collector) throws Exception {
//每来一条数据，存入List中，并注册定时器（只有触发时间一样，定时器就是同一个）
listState.add(value);
context.timerService().registerEventTimeTimer(value.getWindowEnd());
}
//定时器触发时的逻辑
@Override
public void onTimer(long timestamp, onTimerContext ctx, Collector out) throws Exception {
//转换成Arraylist再排序
ArrayList itemViewCounts = Lists.newArrayList(listState.get().iterator());
itemViewCounts.sort(new Comparator() {
@Override
public int compare(ItemViewCount o1, ItemViewCount o2) {//前-后小于0，为倒序
if(o1.getCount() > o2.getCount())
return -1;
else if (o1.getCount() == o2.getCount())
return 0;
else
return 1;
}
});
//定义输出结果格式
StringBuilder resultBuilder=new StringBuilder();
resultBuilder.append("===================n");
resultBuilder.append("窗口结束时间：").append(new Timestamp(timestamp)).append("n"); //输出windowend
//遍历输出
for (int i = 0; i < Math.min(topSize,itemViewCounts.size()); i++) {
ItemViewCount currentItemViewCount = itemViewCounts.get(i);
resultBuilder.append("Number").append(i+1).append(":")
.append("商品ID：").append(currentItemViewCount.getItemId())
.append("浏览量：").append(currentItemViewCount.getCount())
.append("n");
}
resultBuilder.append("===================nn");
Thread.sleep(1000L);//控制输出频率
out.collect(resultBuilder.toString());
listState.clear();//清空状态
}
@Override
public void close() throws Exception {
listState.clear();//清空状态
}
}
}
这里消费kafka的时候setStreamTimeCharacteristic为TimeCharacteristic.EventTime
- flink的TimeCharacteristic枚举定义了三类值，分别是ProcessingTime、IngestionTime、EventTime
- ProcessingTime是以operator处理的时间为准，它使用的是机器的系统时间来作为data stream的时间；IngestionTime是以数据进入flink streaming data flow的时间为准；EventTime是以数据自带的时间戳字段为准，应用程序需要指定如何从record中抽取时间戳字段
- 指定为EventTime的source需要自己定义event time以及emit watermark，或者在source之外通过assignTimestampsAndWatermarks在程序手工指定
