1. 环境
kafka：1.0.0；
Flink：1.11.0；
2. 发送测试数据
首先发送数据到kafka测试集群，maven依赖：
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.4.1</version> </dependency>发送代码：
importcom.alibaba.fastjson.JSON; importcom.alibaba.fastjson.JSONObject; importjodd.util.ThreadUtil; importorg.apache.commons.lang3.StringUtils; importorg.junit.Test; importjava.io.*; publicclassSendDataToKafka{ @Test publicvoidsendData()throwsIOException{ Stringinpath="E:\\我的文件\\click.txt"; Stringtopic="click_test"; intcnt=0; Stringline; InputStreaminputStream=newFileInputStream(inpath); Readerreader=newInputStreamReader(inputStream); LineNumberReaderlnr=newLineNumberReader(reader); while((line=lnr.readLine())!=null){ //这里的KafkaUtil它是生产者和消费者的工具，可自行实现 KafkaUtil.sendDataToKafka(topic,String.valueOf(cnt),line); cnt=cnt 1; ThreadUtil.sleep(100); } } }3. 主要程序
先定义个pojo：
@NoArgsConstructor @AllArgsConstructor @Data @ToString publicclassUserClickModel{ privateStringdate; privateStringproduct; privateintuid; privateintpv; privateintuv; }然后是使用Flink消费kafka，指定Watermark，通过KeyBy分流，进入滚动窗口通过状态保存函数pv和uv。
publicclassUserClickMain{ privatestaticfinalMap<String,String>config=nbsp;Configuration.initConfig("commons.xml");
public static void main(String[] args) throws Exception {
// 初始化环境，配置相关属性
StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
senv.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
senv.setStateBackend(new FsStateBackend("hdfs://bigdata/flink/checkpoints/userClick"));
// 读取kafka
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", config.get("kafka-ipport"));
kafkaProps.setProperty("group.id", config.get("kafka-groupid"));
// kafkaProps.setProperty("auto.offset.reset", "earliest");
// watrmark 允许数据延迟时间
long maxOutOfOrderness = 5 * 1000L;
SingleOutputStreamOperator<UserClickModel> dataStream = senv.addSource(
new FlinkKafkaConsumer<>(
config.get("kafka-topic"),
new SimpleStringSchema(),
kafkaProps
))
//设置watermark
.assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofMillis(maxOutOfOrderness))
.withTimestampAssigner((element, recordTimestamp) -> {
// 时间戳须为毫秒
return Long.valueOf(JSON.parseObject(element).getString("timestamp")) * 1000;
})).map(new FCClickMapFunction()).returns(TypeInformation.of(new TypeHint<UserClickModel>() {
}));
// 按照 (date, product) 分组
dataStream.keyBy(new KeySelector<UserClickModel, Tuple2<String, String>>() {
@Override
public Tuple2<String, String> getKey(UserClickModel value) throws Exception {
return Tuple2.of(value.getDate(), value.getProduct());
}
})
// 一天为窗口，指定时间起点比时间戳时间早8个小时
.window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
// 10s触发一次计算，更新统计结果
.trigger(ContinuousEventTimeTrigger.of(Time.seconds(10)))
// 计算pv uv
.process(new MyProcessWindowFunctionBitMap())
// 保存结果到mysql
.addSink(new FCClickSinkFunction());
senv.execute(UserClickMain.class.getSimpleName());
}
}代码都是一些常规代码，但是还是有几点需要注意的。
-
设置watermark，flink1.11中使用WatermarkStrategy，老的已经废弃了；
-
我的数据里面时间戳是秒，需要乘以1000，flink提取时间字段，必须为
毫秒； -
.window只传入一个参数，表明是滚动窗口，TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8))这里指定了窗口的大小为一天，由于中国北京时间是东8区，比国际时间早8个小时，需要引入offset，可以自行进入该方法源码查看英文注释。
Rather than that,if you are living in somewhere which is not using UTC±00:00 time,
* such as China which is using UTC+08:00,and you want a time window with size of one day,
* and window begins at every 00:00:00 of local time,you may use {@code of(Time.days(1),Time.hours(-8))}.
* The parameter of offset is {@code Time.hours(-8))} since UTC+08:00 is 8 hours earlier than UTC time.-
一天大小的窗口，根据
watermark机制一天触发计算一次，显然是不合理的，需要用trigger函数指定触发间隔为10s一次，这样我们的pv和uv就是10s更新一次结果。
4. 关键代码，计算uv
由于这里用户id刚好是数字，可以使用bitmap去重，简单原理是：把 user_id 作为 bit 的偏移量 offset，设置为 1 表示有访问，使用 1 MB的空间就可以存放 800 多万用户的一天访问计数情况。
redis是自带bit数据结构的，不过为了尽量少依赖外部存储媒介，这里自己实现bit，引入相应maven依赖即可：
<dependency>
<groupId>org.roaringbitmap</groupId>
<artifactId>RoaringBitmap</artifactId>
<version>0.8.0</version>
</dependency>计算pv、uv的代码其实都是通用的，可以根据自己的实际业务情况快速修改的：
public class MyProcessWindowFunctionBitMap extends ProcessWindowFunction<UserClickModel, UserClickModel, Tuple<String, String>, TimeWindow> {
private transient ValueState<Integer> pvState;
private transient ValueState<Roaring64NavigableMap> bitMapState;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
ValueStateDescriptor<Integer> pvStateDescriptor = new ValueStateDescriptor<>("pv", Integer.class);
ValueStateDescriptor<Roaring64NavigableMap> bitMapStateDescriptor = new ValueStateDescriptor("bitMap"
, TypeInformation.of(new TypeHint<Roaring64NavigableMap>() {}));
// 过期状态清除
StateTtlConfig stateTtlConfig = StateTtlConfig
.newBuilder(Time.days(1))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build();
// 开启ttl
pvStateDescriptor.enableTimeToLive(stateTtlConfig);
bitMapStateDescriptor.enableTimeToLive(stateTtlConfig);
pvState = this.getRuntimeContext().getState(pvStateDescriptor);
bitMapState = this.getRuntimeContext().getState(bitMapStateDescriptor);
}
@Override
public void process(Tuple2<String, String> key, Context context, Iterable<UserClickModel> elements, Collector<UserClickModel> out) throws Exception {
// 当前状态的pv uv
Integer pv = pvState.value();
Roaring64NavigableMap bitMap = bitMapState.value();
if(bitMap == null){
bitMap = new Roaring64NavigableMap();
pv = 0;
}
Iterator<UserClickModel> iterator = elements.iterator();
while (iterator.hasNext()){
pv = pv + 1;
int uid = iterator.next().getUid();
//如果userId可以转成long
bitMap.add(uid);
}
// 更新pv
pvState.update(pv);
UserClickModel UserClickModel = new UserClickModel();
UserClickModel.setDate(key.f0);
UserClickModel.setProduct(key.f1);
UserClickModel.setPv(pv);
UserClickModel.setUv(bitMap.getIntCardinality());
out.collect(UserClickModel);
}
}-
由于计算
uv第二天的时候，就不需要第一天数据了，要及时清理内存中前一天的状态，通过ttl机制过期； -
最终结果保存到mysql里面，如果数据结果分类聚合太多，要注意
mysql压力，这块可以自行优化；
五、其它方法
除了使用bitmap去重外，还可以使用Flink SQL,编码更简洁，还可以借助外面的媒介Redis去重：
-
基于 set
-
基于 bit
-
基于 HyperLogLog
-
基于bloomfilter
具体思路是，计算pv、uv都塞入redis里面，然后再获取值保存统计结果，也是比较常用的。
HDFS的快照讲解Hadoop 数据迁移用法详解Hbase修复工具Hbck数仓建模分层理论一文搞懂Hive的数据存储与压缩大数据组件重点学习这几个