[[432405]]

PV(訪問量)：即Page View, 即頁面瀏覽量或點(diǎn)擊量，用戶每次刷新即被計(jì)算一次。

UV(獨(dú)立訪客)：即Unique Visitor,訪問您網(wǎng)站的一臺(tái)電腦客戶端為一個(gè)訪客。00:00-24:00內(nèi)相同的客戶端只被計(jì)算一次。

計(jì)算網(wǎng)站App的實(shí)時(shí)pv和uv，是很常見的統(tǒng)計(jì)需求，這里提供通用的計(jì)算方法，不同的業(yè)務(wù)需求只需要小改即可拿來即用。

需求

利用Flink實(shí)時(shí)統(tǒng)計(jì)，從0點(diǎn)到當(dāng)前的pv、uv。

一、需求分析

從Kafka發(fā)送過來的數(shù)據(jù)含有：時(shí)間戳、時(shí)間、維度、用戶id，需要從不同維度統(tǒng)計(jì)從0點(diǎn)到當(dāng)前時(shí)間的pv和uv，第二天0點(diǎn)重新開始計(jì)數(shù)第二天的。

二、技術(shù)方案

Kafka數(shù)據(jù)可能會(huì)有延遲亂序，這里引入watermark;

通過keyBy分流進(jìn)不同的滾動(dòng)window，每個(gè)窗口內(nèi)計(jì)算pv、uv;

由于需要保存一天的狀態(tài)，process里面使用ValueState保存pv、uv;

使用BitMap類型ValueState，占內(nèi)存很小，引入支持bitmap的依賴;

保存狀態(tài)需要設(shè)置ttl過期時(shí)間，第二天把第一天的過期，避免內(nèi)存占用過大。

三、數(shù)據(jù)準(zhǔn)備

這里假設(shè)是用戶訂單數(shù)據(jù)，數(shù)據(jù)格式如下：

{"time":"2021-10-31 22:00:01","timestamp":"1635228001","product":"蘋果手機(jī)","uid":255420} 
{"time":"2021-10-31 22:00:02","timestamp":"1635228001","product":"MacBook Pro","uid":255421} 

四、代碼實(shí)現(xiàn)

整個(gè)工程代碼截圖如下(抹去了一些不方便公開的信息)：

pvuv-project

1. 環(huán)境

kafka：1.0.0;

Flink：1.11.0;

2. 發(fā)送測試數(shù)據(jù)

首先發(fā)送數(shù)據(jù)到kafka測試集群，maven依賴：

<dependency> 
    <groupId>org.apache.kafka</groupId> 
    <artifactId>kafka-clients</artifactId> 
    <version>2.4.1</version> 
</dependency> 

2.4.1

發(fā)送代碼：

import com.alibaba.fastjson.JSON; 
import com.alibaba.fastjson.JSONObject; 
import jodd.util.ThreadUtil; 
import org.apache.commons.lang3.StringUtils; 
import org.junit.Test; 
 
import java.io.*; 
 
public class SendDataToKafka { 
 
    @Test 
    public void sendData() throws IOException { 
        String inpath = "E:\\我的文件\\click.txt"; 
        String topic = "click_test"; 
        int cnt = 0; 
        String line; 
        InputStream inputStream = new FileInputStream(inpath); 
        Reader reader = new InputStreamReader(inputStream); 
        LineNumberReader lnr = new LineNumberReader(reader); 
        while ((line = lnr.readLine()) != null) { 
            // 這里的KafkaUtil是個(gè)生產(chǎn)者、消費(fèi)者工具類，可以自行實(shí)現(xiàn) 
            KafkaUtil.sendDataToKafka(topic, String.valueOf(cnt), line); 
            cnt = cnt + 1; 
            ThreadUtil.sleep(100); 
        } 
    } 
} 

3. 主要程序

先定義個(gè)pojo：

@NoArgsConstructor 
@AllArgsConstructor 
@Data 
@ToString 
public class UserClickModel { 
    private String date; 
    private String product; 
    private int uid; 
    private int pv; 
    private int uv; 
} 

接著就是使用Flink消費(fèi)kafka，指定Watermark，通過KeyBy分流，進(jìn)入滾動(dòng)窗口函數(shù)通過狀態(tài)保存pv和uv。

public class UserClickMain { 
 
    private static final Map<String, String> config = Configuration.initConfig("commons.xml"); 
 
    public static void main(String[] args) throws Exception { 
 
        // 初始化環(huán)境，配置相關(guān)屬性 
        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 允許數(shù)據(jù)延遲時(shí)間 
        long maxOutOfOrderness = 5 * 1000L; 
        SingleOutputStreamOperator<UserClickModel> dataStream = senv.addSource( 
                new FlinkKafkaConsumer<>( 
                        config.get("kafka-topic"), 
                        new SimpleStringSchema(), 
                        kafkaProps 
                )) 
                //設(shè)置watermark 
                .assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofMillis(maxOutOfOrderness)) 
                        .withTimestampAssigner((element, recordTimestamp) -> { 
                            // 時(shí)間戳須為毫秒 
                            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()); 
            } 
        }) 
                // 一天為窗口，指定時(shí)間起點(diǎn)比時(shí)間戳?xí)r間早8個(gè)小時(shí) 
                .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8))) 
                // 10s觸發(fā)一次計(jì)算，更新統(tǒng)計(jì)結(jié)果 
                .trigger(ContinuousEventTimeTrigger.of(Time.seconds(10))) 
                // 計(jì)算pv uv 
                .process(new MyProcessWindowFunctionBitMap()) 
                // 保存結(jié)果到mysql 
                .addSink(new FCClickSinkFunction()); 
 
        senv.execute(UserClickMain.class.getSimpleName()); 
    } 
} 

代碼都是一些常規(guī)代碼，但是還是有幾點(diǎn)需要注意的。

注意

設(shè)置watermark，flink1.11中使用WatermarkStrategy，老的已經(jīng)廢棄了;

我的數(shù)據(jù)里面時(shí)間戳是秒，需要乘以1000，flink提取時(shí)間字段，必須為毫秒;

.window只傳入一個(gè)參數(shù)，表明是滾動(dòng)窗口，TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8))這里指定了窗口的大小為一天，由于中國北京時(shí)間是東8區(qū)，比國際時(shí)間早8個(gè)小時(shí)，需要引入offset，可以自行進(jìn)入該方法源碼查看英文注釋。

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.

一天大小的窗口，根據(jù)watermark機(jī)制一天觸發(fā)計(jì)算一次，顯然是不合理的，需要用trigger函數(shù)指定觸發(fā)間隔為10s一次，這樣我們的pv和uv就是10s更新一次結(jié)果。

4. 關(guān)鍵代碼，計(jì)算uv

由于這里用戶id剛好是數(shù)字，可以使用bitmap去重，簡單原理是：把 user_id 作為 bit 的偏移量 offset，設(shè)置為 1 表示有訪問，使用 1 MB的空間就可以存放 800 多萬用戶的一天訪問計(jì)數(shù)情況。

redis是自帶bit數(shù)據(jù)結(jié)構(gòu)的，不過為了盡量少依賴外部存儲(chǔ)媒介，這里自己實(shí)現(xiàn)bit，引入相應(yīng)maven依賴即可：

<dependency> 
    <groupId>org.roaringbitmap</groupId> 
    <artifactId>RoaringBitmap</artifactId> 
    <version>0.8.0</version> 
</dependency> 

計(jì)算pv、uv的代碼其實(shí)都是通用的，可以根據(jù)自己的實(shí)際業(yè)務(wù)情況快速修改的：

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>() {})); 
 
        // 過期狀態(tài)清除 
        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 { 
 
        // 當(dāng)前狀態(tài)的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可以轉(zhuǎn)成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); 
    } 
} 

注意

由于計(jì)算uv第二天的時(shí)候，就不需要第一天數(shù)據(jù)了，要及時(shí)清理內(nèi)存中前一天的狀態(tài)，通過ttl機(jī)制過期;

最終結(jié)果保存到mysql里面，如果數(shù)據(jù)結(jié)果分類聚合太多，要注意mysql壓力，這塊可以自行優(yōu)化;

五、其它方法

除了使用bitmap去重外，還可以使用Flink SQL,編碼更簡潔，還可以借助外面的媒介Redis去重：

• 基于 set
• 基于 bit
• 基于 HyperLogLog
• 基于bloomfilter

具體思路是，計(jì)算pv、uv都塞入redis里面，然后再獲取值保存統(tǒng)計(jì)結(jié)果，也是比較常用的。