Flink's general method for calculating Pv and Uv

[[432405]]

PV (Visit Volume): Page View, which is the number of page views or clicks. It is calculated every time the user refreshes the page.

UV (Unique Visitor): A computer client that visits your website is a visitor. The same client is only counted once between 00:00 and 24:00.

Calculating the real-time PV and UV of website apps is a very common statistical requirement. Here we provide a general calculation method that can be used directly with only minor modifications to meet different business needs.

need

Use Flink to perform real-time statistics from 0 o'clock to the current pv and uv.

1. Demand Analysis

The data sent from Kafka contains: timestamp, time, dimension, and user id. It is necessary to count the pv and uv from 0:00 to the current time from different dimensions, and start counting the next day again at 0:00 the next day.

2. Technical Solution

Kafka data may be delayed and out of order, so watermark is introduced here;

Divide the data into different scrolling windows through keyBy, and calculate pv and uv in each window;

Since the state of one day needs to be saved, ValueState is used in the process to save pv and uv;

Use BitMap type ValueState, which takes up very little memory and introduces dependencies that support bitmap;

To save the status, you need to set the ttl expiration time, and expire the first day's status on the second day to avoid excessive memory usage.

3. Data Preparation

Here we assume that it is user order data, and the data format is as follows:

 { "time" : "2021-10-31 22:00:01" , "timestamp" : "1635228001" , "product" : "Apple phone" , "uid" :255420}
 { "time" : "2021-10-31 22:00:02" , "timestamp" : "1635228001" , "product" : "MacBook Pro" , "uid" :255421}

4. Code Implementation

The screenshot of the entire project code is as follows (some information that is not convenient for public disclosure is erased):

pvuv-project

1. Environment

kafka:1.0.0;

Flink: 1.11.0;

2. Send test data

First, send data to the Kafka test cluster, Maven dependencies:

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

2.4.1

Send code:

 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:\\My File\\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 ) {
 // The KafkaUtil here is a producer and consumer tool class that can be implemented by yourself
            KafkaUtil.sendDataToKafka(topic, String.valueOf(cnt), line);
 cnt = cnt + 1;
            ThreadUtil.sleep(100);
 }
 }
 }

3. Main procedures

First define a pojo:

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

The next step is to use Flink to consume Kafka, specify Watermark, split the data using KeyBy, and enter the rolling window function to save pv and uv through the state.

 public class UserClickMain { 
 
    private static final Map<String, String> config = Configuration.initConfig( "commons.xml" ); 
 
 public   static void main(String[] args) throws Exception { 
 
 // Initialize the environment and configure related properties
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        senv.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        senv.setStateBackend(new FsStateBackend( "hdfs://bigdata/flink/checkpoints/userClick" )); 
 
 // Read 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 allows data delay time
        long maxOutOfOrderness = 5 * 1000L;
        SingleOutputStreamOperator<UserClickModel> dataStream = senv.addSource(
                new FlinkKafkaConsumer<>(
                        config.get( "kafka-topic" ),
                        new SimpleStringSchema(),
                        kafkaProps
 ))
 //Set watermark
                .assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofMillis(maxOutOfOrderness))
                        .withTimestampAssigner((element, recordTimestamp) -> {
 // Timestamp must be in milliseconds
 return Long.valueOf(JSON.parseObject(element).getString( "timestamp" )) * 1000;
                        })).map(new FCClickMapFunction()). returns (TypeInformation. of (new TypeHint<UserClickModel>() {
 })); 
 
 // Group by ( 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());
 }
 })
 // One day is the window, and the specified time starting point is 8 hours earlier than the timestamp time
                .window(TumblingEventTimeWindows. of ( Time .days(1), Time .hours(-8)))
 // Trigger calculation once every 10 seconds and update statistical results
                . trigger (ContinuousEventTimeTrigger. of ( Time .seconds(10)))
 // Calculate pv uv
                .process(new MyProcessWindowFunctionBitMap())
 // Save the results to mysql
                .addSink(new FCClickSinkFunction()); 
 
        senv.execute (UserClickMain.class.getSimpleName());
 }
 }

The code is pretty standard, but there are a few things to note.

Notice

Set watermark. Use WatermarkStrategy in flink1.11. The old one has been deprecated.

The timestamp in my data is in seconds, so I need to multiply it by 1000. Flink extracts the time field, which must be in milliseconds.

.window only passes in one parameter, indicating that it is a tumbling window, TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)). Here, the window size is specified as one day. Since Beijing time in China is Eastern Time Zone 8, which is 8 hours earlier than the international time, an offset needs to be introduced. You can enter the source code of this method to view the English comments.

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.

It is obviously unreasonable to trigger the calculation once a day for a window of one day according to the watermark mechanism. We need to use the trigger function to specify the trigger interval as 10s, so that our pv and uv are updated every 10s.

4. Key code, calculate uv

Since the user ID here is just a number, bitmap can be used for deduplication. The simple principle is: use user_id as the offset of the bit, set it to 1 to indicate access, and use 1 MB of space to store the daily access counts of more than 8 million users.

Redis comes with its own bit data structure, but in order to rely as little as possible on external storage media, we implement bit ourselves here and introduce the corresponding Maven dependency:

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

The codes for calculating pv and uv are actually universal and can be quickly modified according to your actual business situation:

 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>() {})); 
 
 // Clear expired status
        StateTtlConfig stateTtlConfig = StateTtlConfig
                .newBuilder( Time .days(1))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
 .build();
 // Enable 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 { 
 
 // Current state 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 ();
 //If userId can be converted to long
            bitMap.add (uid);
 } 
 
 // Update 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);
 }
 }

Notice

Since the first day's data is no longer needed when calculating UV for the second day, the previous day's status in the memory must be cleared in time and expired through the TTL mechanism;

The final results are saved in MySQL. If there are too many data result classifications and aggregations, pay attention to the pressure on MySQL, which can be optimized by yourself;

5. Other methods

In addition to using bitmap for deduplication, you can also use Flink SQL, which has simpler coding, and you can also use external media Redis for deduplication:

• Set-based
• Based on bit
• Based on HyperLogLog
• Based on bloomfilter

The specific idea is to calculate pv and uv and put them into redis, and then get the values ​​to save the statistical results, which is also commonly used.