Flink计算pv和uv的通用方法

大数据技术派 2023-03-28 原文

关注公众号：大数据技术派，回复“资料”，领取1000G资料。本文发于我的个人博客：Flink计算pv和uv的通用方法

PV(访问量)：即Page View, 即页面浏览量或点击量，用户每次刷新即被计算一次。

UV(独立访客)：即Unique Visitor,访问您网站的一台电脑客户端为一个访客。00:00-24:00内相同的客户端只被计算一次。

计算网站App的实时pv和uv，是很常见的统计需求，这里提供通用的计算方法，不同的业务需求只需要小改即可拿来即用。

需求

利用Flink实时统计，从0点到当前的pv、uv。

一、需求分析

从Kafka发送过来的数据含有：时间戳、时间、维度、用户id，需要从不同维度统计从0点到当前时间的pv和uv，第二天0点重新开始计数第二天的。

二、技术方案

Kafka数据可能会有延迟乱序，这里引入watermark；
通过keyBy分流进不同的滚动window，每个窗口内计算pv、uv；
由于需要保存一天的状态，process里面使用ValueState保存pv、uv；
使用BitMap类型ValueState，占内存很小，引入支持bitmap的依赖；
保存状态需要设置ttl过期时间，第二天把第一天的过期，避免内存占用过大。

三、数据准备

这里假设是用户订单数据，数据格式如下：

{"time":"2021-10-31 22:00:01","timestamp":"1635228001","product":"苹果手机","uid":255420}
{"time":"2021-10-31 22:00:02","timestamp":"1635228001","product":"MacBook Pro","uid":255421}

四、代码实现

整个工程代码截图如下（抹去了一些不方便公开的信息）：

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>

发送代码：

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是个生产者、消费者工具类，可以自行实现
            KafkaUtil.sendDataToKafka(topic, String.valueOf(cnt), line);
            cnt = cnt + 1;
            ThreadUtil.sleep(100);
        }
    }
}

3. 主要程序

先定义个pojo：

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

接着就是使用Flink消费kafka，指定Watermark，通过KeyBy分流，进入滚动窗口函数通过状态保存pv和uv。

public class UserClickMain {

    private static final Map<String, String> config = 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里面，然后再获取值保存统计结果，也是比较常用的。

Flink 计算 code quot UserClickModel FlinkSpark 大数据

有关Flink计算pv和uv的通用方法的更多相关文章

ruby - 如何使用 Nokogiri 的 xpath 和 at_xpath 方法 - 2
我正在学习如何使用Nokogiri，根据这段代码我遇到了一些问题:require'rubygems'require'mechanize'post_agent=WWW::Mechanize.newpost_page=post_agent.get('http://www.vbulletin.org/forum/showthread.php?t=230708')puts"\nabsolutepathwithtbodygivesnil"putspost_page.parser.xpath('/html/body/div/div/div/div/div/table/tbody/tr/td/div
ruby - 如何从 ruby 中的字符串运行任意对象方法？ - 2
总的来说，我对ruby还比较陌生，我正在为我正在创建的对象编写一些rspec测试用例。许多测试用例都非常基础，我只是想确保正确填充和返回值。我想知道是否有办法使用循环结构来执行此操作。不必为我要测试的每个方法都设置一个assertEquals。例如:describeitem,"TestingtheItem"doit"willhaveanullvaluetostart"doitem=Item.new#HereIcoulddotheitem.name.shouldbe_nil#thenIcoulddoitem.category.shouldbe_nilendend但我想要一些方法来使用
ruby - 为什么我可以在 Ruby 中使用 Object#send 访问私有(private)/ protected 方法？ - 2
类classAprivatedeffooputs:fooendpublicdefbarputs:barendprivatedefzimputs:zimendprotecteddefdibputs:dibendendA的实例a=A.new测试a.foorescueputs:faila.barrescueputs:faila.zimrescueputs:faila.dibrescueputs:faila.gazrescueputs:fail测试输出failbarfailfailfail.发送测试[:foo,:bar,:zim,:dib,:gaz].each{|m|a.send(m)resc
ruby - Facter::Util::Uptime:Module 的未定义方法 get_uptime (NoMethodError) - 2
我正在尝试设置一个puppet节点，但rubygems似乎不正常。如果我通过它自己的二进制文件(/usr/lib/ruby/gems/1.8/gems/facter-1.5.8/bin/facter)在cli上运行facter，它工作正常，但如果我通过由rubygems(/usr/bin/facter)安装的二进制文件，它抛出:/usr/lib/ruby/1.8/facter/uptime.rb:11:undefinedmethod`get_uptime'forFacter::Util::Uptime:Module(NoMethodError)from/usr/lib/ruby
Ruby 方法() 方法 - 2
我想了解Ruby方法methods()是如何工作的。我尝试使用“ruby方法”在Google上搜索，但这不是我需要的。我也看过ruby-doc.org，但我没有找到这种方法。你能详细解释一下它是如何工作的或者给我一个链接吗？更新我用methods()方法做了实验，得到了这样的结果:'labrat'代码classFirstdeffirst_instance_mymethodenddefself.first_class_mymethodendendclassSecond使用类#returnsavailablemethodslistforclassandancestorsputsSeco
ruby-on-rails - Rails 3.2.1 中 ActionMailer 中的未定义方法 'default_content_type=' - 2
我在我的项目中添加了一个系统来重置用户密码并通过电子邮件将密码发送给他，以防他忘记密码。昨天它运行良好(当我实现它时)。当我今天尝试启动服务器时，出现以下错误。=>BootingWEBrick=>Rails3.2.1applicationstartingindevelopmentonhttp://0.0.0.0:3000=>Callwith-dtodetach=>Ctrl-CtoshutdownserverExiting/Users/vinayshenoy/.rvm/gems/ruby-1.9.3-p0/gems/actionmailer-3.2.1/lib/action_mailer
ruby - Highline 询问方法不会使用同一行 - 2
设置:狂欢ruby1.9.2高线(1.6.13)描述:我已经相当习惯在其他一些项目中使用highline，但已经有几个月没有使用它了。现在，在Ruby1.9.2上全新安装时，它似乎不允许在同一行回答提示。所以以前我会看到类似的东西:require"highline/import"ask"Whatisyourfavoritecolor?"并得到:Whatisyourfavoritecolor?|现在我看到类似的东西:Whatisyourfavoritecolor?|竖线(|)符号是我的终端光标。知道为什么会发生这种变化吗？最佳答案
ruby - 主要 :Object when running build from sublime 的未定义方法 `require_relative' - 2
我已经从我的命令行中获得了一切，所以我可以运行rubymyfile并且它可以正常工作。但是当我尝试从sublime中运行它时，我得到了undefinedmethod`require_relative'formain:Object有人知道我的sublime设置中缺少什么吗？我正在使用OSX并安装了rvm。最佳答案或者，您可以只使用“require”，它应该可以正常工作。我认为“require_relative”仅适用于ruby1.9+ 关于ruby-主要:Objectwhenrun
ruby-on-rails - 使用一系列等级计算字母等级 - 2
这里是Ruby新手。完成一些练习后碰壁了。练习:计算一系列成绩的字母等级创建一个方法get_grade来接受测试分数数组。数组中的每个分数应介于0和100之间，其中100是最大分数。计算平均分并将字母等级作为字符串返回，即“A”、“B”、“C”、“D”、“E”或“F”。我一直返回错误:avg.rb:1:syntaxerror,unexpectedtLBRACK,expecting')'defget_grade([100,90,80])^avg.rb:1:syntaxerror,unexpected')',expecting$end这是我目前所拥有的。我想坚持使用下面的方法或.join，
ruby - 多个属性的 update_column 方法 - 2
我有一个具有一些属性的模型:attr1、attr2和attr3。我需要在不执行回调和验证的情况下更新此属性。我找到了update_column方法，但我想同时更新三个属性。我需要这样的东西:update_columns({attr1:val1,attr2:val2,attr3:val3})代替update_column(attr1,val1)update_column(attr2,val2)update_column(attr3,val3) 最佳答案您可以使用update_columns(attr1:val1,attr2:val2