php网站uvpv统计技巧_大年夜数据技能分享运用Spark Streaming SQL进行PVUV统计

1.背景先容

PV/UV统计是流式剖析一个常见的场景。
通过PV可以对访问的网站做流量或热点剖析，例如广告主可以通过PV值预估投放广告网页所带来的流量以及广告收入。
其余一些场景须要对访问的用户作剖析，比如剖析用户的网页点击行为，此时就须要对UV做统计。

利用Spark Streaming SQL，并结合Redis可以很方便进行PV/UV的统计。
本文将先容通过Streaming SQL消费Loghub中存储的用户访问信息，对过去1分钟内的数据进行PV/UV统计，将结果存入Redis中。

2.准备事情

git clone git@github.com:aliyun/aliyun-emapreduce-sdk.git cd aliyun-emapreduce-sdk git checkout -b master-2.x origin/master-2.x mvn 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='interval');

个中，statistic_info为Redis存储结果的表名，interval对应统计结果中的interval字段;配置项${redis_host}的值根据实际配置。

3.2创建流作业

CREATE SCAN loghub_scan ON loghub_source USING STREAM OPTIONS( watermark.column='__time__', watermark.delayThreshold='10 second'); CREATE STREAM job OPTIONS( checkpointLocation=${checkpoint_location}) INSERT INTO redis_sink SELECT COUNT(user_ip) AS pv, approx_count_distinct( user_ip) AS uv, window.end AS interval FROM loghub_scan GROUP BY TUMBLING(__time__, interval 1 minute), window;

4.3查看统计结果

终极的统计结果如下图所示

可以看到，每隔一分钟都会天生一条数据，key的形式为表名:interval，value为pv和uv的值。

3.4实现覆盖更新

将结果表的配置项key.column修正为一个固定的值，例如定义如下

CREATE TABLE redis_sink USING redis OPTIONS( table='statistic_info', host=${redis_host}, key.column='statistic_type');

创建流作业的SQL改为

CREATE STREAM job OPTIONS( checkpointLocation='/tmp/spark-test/checkpoint') INSERT INTO redis_sink SELECT \"大众PV_UV\"大众 as statistic_type,COUNT(user_ip) AS pv, approx_count_distinct( user_ip) AS uv, window.end AS interval FROM loghub_scan GROUP BY TUMBLING(__time__, interval 1 minute), window;

终极的统计结果如下图所示

可以看到，Redis中值保留了一个值，这个值每分钟都被更新，value包含pv、uv和interval的值。

4.总结

本文简要先容了利用Streaming SQL结合Redis实现流式处理中统计PV/UV的需求。