1、访问对象(VisitDto)：
     private String ip;          // 访问IP

    

    private String visitDate; // 访问时间

    

    private String visitMode; //访问方式：post get ...

    

    private String visitUrl;  //访问路径

    

    private String userBroset; //访问用户的浏览器

    

    private String userOs;     //访问用户的操作系统


2、统计方法，并将数据放到缓存中：
public static void counts(HttpServletRequest request){

        try{

            VisitDto dto = new VisitDto();

            String ip = IpUtil.getIpAddr(request);// ip地址

            String key = new SimpleDateFormat("yyyy-MM-dd").format(new Date());

            key = UP_COUNT_KEY + key;

            String date = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date());// 访问时间

            String mode  = request.getMethod();// 访问方式 Post Get

            String agent = request.getHeader("user-agent").toLowerCase();

            //得到用户的浏览器名

            String userbrowser = getBrowserName(agent);

            //得到用户的操作系统名

            String useros = getOsName(agent); 

            String url = request.getRequestURI();// 访问路径

            String id = request.getSession().getId();

            dto.setIp(ip);

            dto.setVisitDate(date);

            dto.setVisitMode(mode);

            dto.setVisitUrl(url);

            dto.setUserBroset(userbrowser);

            dto.setUserOs(useros);

            if(StringUtils.isNotBlank(id)){

                JSONObject jsonObject = JSONObject.fromObject(dto);

                mm.put(id, jsonObject.toString());

                JedisSessionClient.hmset(key, mm, (int) TimeUnit.DAYS.toSeconds(30));

            }

        }catch(Exception e){

            log.error(e.getMessage(),e);

        }

    }


 

    /**

      * 获取浏览器版本信息

      * @Title: getBrowserName

      *  >>Browser:Chrome > FF > IE > ... 

      * @param agent

      * @return

      */


    public static String getBrowserName(String agent) {

          if(StringUtils.isBlank(agent)) {  

               return null;  

           } 

          if(agent.indexOf("msie 7")>0){

           return "ie7";

          }else if(agent.indexOf("msie 8")>0){

           return "ie8";

          }else if(agent.indexOf("msie 9")>0){

           return "ie9";

          }else if(agent.indexOf("msie 10")>0){

           return "ie10";

          }else if(agent.indexOf("msie")>0){

           return "ie";

          }else if(agent.indexOf("opera")>0){

           return "opera";

          }else if(agent.indexOf("chrome")>0){

           return "chrome";

          }else if(agent.indexOf("firefox")>0){

           return "firefox";

          }else if(agent.indexOf("webkit")>0){

           return "webkit";

          }else if(agent.indexOf("gecko")>0 && agent.indexOf("rv:11")>0){

           return "ie11";

          }else{

           return "Others";

          }

     }

    

    /**

     * 获取操作系统信息

     * @Title: getOsName

     *  >>操作系统:Windows > 苹果 > 安卓 > Linux > ... 

     * @param agent

     * @return

     */

    public static String getOsName(String userAgent){

       if (StringUtils.isBlank(userAgent)) {  

           return null;  

       }   

       if (userAgent.indexOf("windows")>0) {//主流应用靠前  

               return "Windows";

       } else if (userAgent.indexOf("mac os x") >0) {  

           return "Mac OS X";

       }else if(userAgent.indexOf("android") >0){

           return "Android";

       }else if(userAgent.indexOf("liunx") >0){

           return "Liunx";

       }else{

           return "Others";

       }

   }  

3、缓存方法：


 public class JedisSessionClient {

    private static JedisPool poolMaster;


    private JedisSessionClient() {


    }

  private static Logger log = LoggerFactory.getLogger(JedisSessionClient.class);

    static{

        try {

            Properties p = new Properties();

//                String filePath = new JedisClient().getClass().getResource("/").getPath() + "redisConfig.properties";

//                log.info("filePath:" + filePath);

//                InputStream is = new BufferedInputStream(new FileInputStream(filePath));

            ClassLoader classloader = Thread.currentThread().getContextClassLoader();

            InputStream is = classloader.getResourceAsStream("/redisConfig.properties");

            p.load(is);

            String redisSession = p.getProperty("redisSession").trim();

            // password = p.getProperty("password");

            int redisPort = Integer.parseInt(p.getProperty("redisPort").trim());

            int timeout = Integer.parseInt(p.getProperty("timeout").trim());

            JedisPoolConfig configMaster = new JedisPoolConfig();

            configMaster.setMaxTotal(30000);// 设置最大连接数

            configMaster.setMaxIdle(100); // 设置最大空闲数

            configMaster.setMinIdle(10);

            configMaster.setMaxWaitMillis(10000);// 设置超时时间

            configMaster.setTestWhileIdle(false);

            configMaster.setTestOnBorrow(false);

            poolMaster = new JedisPool(configMaster, redisSession,redisPort, timeout);

            log.info(">>>>>>>>>redisSession ip:" + redisSession);

            log.info(">>>>>>>>>redisSession port:"+redisPort);

            log.info(">>>>>>>>>redisSession timeout:"+timeout);

            log.info(">>>>>>>>>会话管理redis master连接池初始化成功...");

        } catch (Exception e) {

            log.info(">>>>>>>>>会话管理redis master连接池初始化失败，系统无法正常使用！！！");

            e.printStackTrace();

        }

        }

    private static Jedis getJedis() throws Exception {

        return poolMaster.getResource();

    }


    private static void returnResource(Jedis jedis) {

        if (jedis != null) {

            jedis.disconnect();

            poolMaster.returnResourceObject(jedis);

        }

    }

  
    public static void hmset(String key,Map<String,String> map,int seconds) throws Exception {

        if (null == key || "".equals(key.trim())) {

            throw new Exception("the key can not be null.");

        }

        Jedis jedisMaster = null;

        try {

            jedisMaster = getJedis();

            jedisMaster.hmset(key, map);

            if(seconds > 0){

                jedisMaster.expire(key, seconds);                

            }

        } finally {

            returnResource(jedisMaster);

        }

    }
  public static Set<String> hkeys(String key)throws Exception {

        if (null == key || "".equals(key.trim())) {

            throw new Exception("the key can not be null.");

        }

        Jedis jedisMaster = null;

        Set<String> set = null;

        try {

            jedisMaster = getJedis();

            set = jedisMaster.hkeys(key);

        } finally {

            returnResource(jedisMaster);

        }

        return set;

    }

 public static List<String> hvals(String key)throws Exception {

        if (null == key || "".equals(key.trim())) {

            throw new Exception("the key can not be null.");

        }

        Jedis jedisMaster = null;

        List<String> list = null;

        try {

            jedisMaster = getJedis();

            list = jedisMaster.hvals(key);

        } finally {

            returnResource(jedisMaster);

        }

        return list;

    }

}

4、从缓存中获取对应key，value：
List<VisitDto> vList = new ArrayList<VisitDto>();
List<String> list = new ArrayList<String>();

 Set<String> set = JedisSessionClient.hkeys(key);：获取所有key
 if(!set.isEmpty()){

                        for (String str : set) {  

                            VisitDto dto = new VisitDto();

                            dto.setIp(str);

                            dto.setVisitDate(date);

                            vList.add(dto);

                            } 

                    }

  JedisSessionClient.hvals(kye) 获取所有value
  list = JedisSessionClient.hvals(key);

                    if(null !=list && list.size()>0){

                        for(int i=0;i<list.size();i++){

                            VisitDto dto = new VisitDto();

                            JSONObject jsonObj = JSONObject.fromObject(list.get(i));

                            dto = (VisitDto)JSONObject.toBean(jsonObj,VisitDto.class);

                            vList.add(dto);

                            }

                        }

1.背景介绍
PV/UV统计是流式分析一个常见的场景。通过PV可以对访问的网站做流量或热点分析，例如广告主可以通过PV值预估投放广告网页所带来的流量以及广告收入。另外一些场景需要对访问的用户作分析，比如分析用户的网页点击行为，此时就需要对UV做统计。
使用Spark Streaming SQL，并结合Redis可以很方便进行PV/UV的统计。本文将介绍通过Streaming SQL消费Loghub中存储的用户访问信息，对过去1分钟内的数据进行PV/UV统计，将结果存入Redis中。
2.准备工作
创建E-MapReduce 3.23.0以上版本的Hadoop集群。
下载并编译E-MapReduce-SDK包
git clone git@github.com:aliyun/aliyun-emapreduce-sdk.gitcd aliyun-emapreduce-sdkgit checkout -b master-2.x origin/master-2.xmvn clean package -DskipTests
编译完后, assembly/target目录下会生成emr-datasources_shaded_${version}.jar，其中${version}为sdk的版本。
数据源
本文采用Loghub作为数据源，有关日志采集、日志解析请参考日志服务。
3.统计PV/UV
一般场景下需要将统计出的PV/UV以及相应的统计时间存入Redis。其他一些业务场景中，也会只保存最新结果，用新的结果不断覆盖更新旧的数据。以下首先介绍第一种情况的操作流程。
3.1启动客户端
命令行启动streaming-sql客户端
streaming-sql --master yarn-client --num-executors 2 --executor-memory 2g --executor-cores 2 --jars emr-datasources_shaded_2.11-${version}.jar --driver-class-path emr-datasources_shaded_2.11-${version}.jar
也可以创建SQL语句文件，通过streaming-sql -f的方式运行。
3.1定义数据表
数据源表定义如下
CREATE TABLE loghub_source(user_ip STRING, __time__ TIMESTAMP) USING loghub OPTIONS(sls.project=${sls.project},sls.store=${sls.store},access.key.id=${access.key.id},access.key.secret=${access.key.secret},endpoint=${endpoint});
其中，数据源表包含user_ip和__time__两个字段，分别代表用户的IP地址和loghub上的时间列。OPTIONS中配置项的值根据实际配置。
结果表定义如下
CREATE TABLE redis_sink USING redis OPTIONS(table='statistic_info',host=${redis_host},key.column='user_ip');
其中，user_ip对应数据中的用户IP字段，配置项${redis_host}的值根据实际配置。
3.2创建流作业
CREATE SCAN loghub_scanON loghub_sourceUSING STREAMOPTIONS(watermark.column='__time__',watermark.delayThreshold='10 second');CREATE STREAM jobOPTIONS(checkpointLocation=${checkpoint_location})INSERT INTO redis_sinkSELECT COUNT(user_ip) AS pv, approx_count_distinct( user_ip) AS uv, window.end AS intervalFROM loghub_scanGROUP BY TUMBLING(__time__, interval 1 minute), window;
4.3查看统计结果
最终的统计结果如下图所示
可以看到，每隔一分钟都会生成一条数据，key的形式为表名:interval，value为pv和uv的值。
3.4实现覆盖更新
将结果表的配置项key.column修改为一个固定的值，例如定义如下
CREATE TABLE redis_sinkUSING redis OPTIONS(table='statistic_info',host=${redis_host},key.column='statistic_type');
创建流作业的SQL改为
CREATE STREAM jobOPTIONS(checkpointLocation='/tmp/spark-test/checkpoint')INSERT INTO redis_sinkSELECT "PV_UV" as statistic_type,COUNT(user_ip) AS pv, approx_count_distinct( user_ip) AS uv, window.end AS intervalFROM loghub_scanGROUP BY TUMBLING(__time__, interval 1 minute), window;
最终的统计结果如下图所示
可以看到，Redis中值保留了一个值，这个值每分钟都被更新，value包含pv、uv和interval的值。
4.总结
本文简要介绍了使用Streaming SQL结合Redis实现流式处理中统计PV/UV的需求。后续文章，我将介绍Spark Streaming SQL的更多内容
作者：关文选，花名云魄，阿里云E-MapReduce 高级开发工程师，专注于流式计算，Spark Contributor
本文为云栖社区内容，未经允许不得转载。

一、背景
在考拉先生公司做的第一个项目是进行统计pvuv。通过一定的规则提取nginx日志文件来进行统计。

二、第一个版本
一开始做的是统计唯品会的pvuv统计，观察流量。主要设计图如下：

很简单，将每天唯品会的广告的pvuv统计好，然后写进excel。然后邮件定时发送给需求人员。

三、第二个版本
统计完唯品会之后，接到新的需求是统计小程序的pvuv。不废话，上流程图。

跟第一个版本不同的是，所有的统计数据会先插入数据库。然后另一个脚本读取数据库数据写进excel发送邮件。
为什么要写进数据库。需求需要，需求要求要时间段的统计，比如一个月或者一周的统计，为了方便操作，所以使用
数据库操作。另一方面是，以后可能会做成一个系统，所以入库。

四、第三个版本
做完小程序的pvuv后，又来了新的pvuv统计需求。这时，上头提出了一个新的方案。

因为当时的做法是，每次来一个新的pvuv统计，都要编写python脚本。现在改成将日志的元数据录入到mongodb里面。
然后配置规则，python脚本进行统计。然后交给业务。这样，假如后面有pvuv或者其他关于日志的处理，直接更改
查询规则就可以了，而不用从新开发。

这次，学习到是是：想问题的时候，一定是从大的方面思考，而不是基于目前的业务。思维角度要开阔。




https://blog.csdn.net/kelindame/article/details/73072603

文章目录
1 概述
2 基本使用


本文学习在 Redis 中通过 HyperLogLog 实现 UV 统计。


1 概述
一般我们评估一个网站的访问量，有几个主要的参数：PV（Page View）网页的浏览量、UV（User View）访问的用户量。有很多第三方工具可以统计，如 cnzz，友盟等。如果自己实现的话，PV 比较简单，可以直接通过 Redis 计数器实现。但是 UV 就不一样，UV 涉及到去重的问题。
常规思路：我们首先需要在前端给每一个用户生成一个唯一 id，无论是登录用户还是未登录用户都需要，这个 id 伴随着请求一起到达后端，在后端我们可以通过 Set 数据结构中的 sadd 命令来存储这个 id，最后通过 scard 统计集合大小，进而得出 UV 数据。
按上述思路，如果是千万级别的 UV，需要的存储空间就非常惊人，用 Set 就不是很合适了。一般来说像 UV 统计这种，不需要特别精确，比如 800W 和 803W 的 UV，其实差别不大。因此，我们可以使用 HyperLogLog 来高效的实现。
2 基本使用
Redis 中提供的 HyperLogLog 就是专门用来解决上述问题的，HyperLogLog 提供了一套不怎么精确但是够用的去重方案，会有误差，官方给出的误差数据是 0.81%，这对于 UV 的统计够用了。
HyperLogLog 主要提供了以下命令：

pfadd：用来添加记录，类似于 sadd ，添加过程中，重复的记录会自动去重。
pfcount：则用来统计数据。
pfmerge：合并多个统计结果，在合并的过程中，会自动去重多个集合中重复的元素。

数据量少的时候看不出来误差，我们在 Java 中多添加几个元素：
public class HyperLogLogTest {
    @Test
    public void testHyperLogLog() {
        Redis redis = new Redis();
        redis.execute(jedis -> {
            for (int i = 0; i < 1000; i++) {
                // 重复加入数据，理论值上总数为 1001
                jedis.pfadd("uv", "u" + i, "u" + (i + 1));
            }
            long uv = jedis.pfcount("uv");
            System.out.println(uv);
        });
    }
}

理论值上总数为 1001，实际打印出来 994，有误差，但是在可以接受的范围内。


Redis 教程合集（微信左下方阅读全文可直达）。
本文示例代码：https://github.com/cxy35/samples/tree/master/redis/redis-jedis


扫码关注微信公众号 程序员35 ，获取最新技术干货，畅聊 #程序员的35，35的程序员# 。独立站点：https://cxy35.com