java利用友盟计算pvuv_Flink计算PV，UV的案例及问题分析

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

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

一个UV可以用很多PV&＃xff0c;一个PV也只能对应一个IP

没有这些数据的支持&＃xff0c;意味着你不知道产品的发展情况&＃xff0c;用户获取成本,UV,PV,注册转化率;没有这些数据做参考&＃xff0c;你不会知道接下来提供什么建议给领导采纳&＃xff0c;也推测不出领导为啥烦忧&＃xff0c;那么就么有任何表现的机会。

举两个UV计算的场景&＃xff1a;

1. 实时计算当天零点起&＃xff0c;到当前时间的uv。

2. 实时计算当天每个小时的UV。0点...12点...24点

请问这个用spark streaming如何实现呢&＃xff1f;是不是很难有好的思路呢&＃xff1f;

今天主要是想给大家用flink来实现一下&＃xff0c;在这方面flink确实比较优秀了。

主要技术点就在group by的使用。

下面就是完整的案例&＃xff1a;

package org.table.uv;

import org.apache.flink.api.common.typeinfo.Types;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.TimeCharacteristic;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import org.apache.flink.table.api.Table;

import org.apache.flink.table.api.TableEnvironment;

import org.apache.flink.table.api.java.StreamTableEnvironment;

import org.apache.flink.table.descriptors.Json;

import org.apache.flink.table.descriptors.Kafka;

import org.apache.flink.table.descriptors.Rowtime;

import org.apache.flink.table.descriptors.Schema;

import org.apache.flink.types.Row;

public class ComputeUVDay {

public static void main(String[] args) throws Exception {

StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

StreamTableEnvironment tEnv &＃61; TableEnvironment.getTableEnvironment(env);

tEnv.registerFunction("DateUtil",new DateUtil());

tEnv.connect(

new Kafka()

.version("0.10")

// "0.8", "0.9", "0.10", "0.11", and "universal"

.topic("jsontest")

.property("bootstrap.servers", "localhost:9092")

.property("group.id","test")

.startFromLatest()

)

.withFormat(

new Json()

.failOnMissingField(false)

.deriveSchema()

)

.withSchema(

new Schema()

.field("rowtime", Types.SQL_TIMESTAMP)

.rowtime(new Rowtime()

.timestampsFromField("eventtime")

.watermarksPeriodicBounded(2000)

)

.field("fruit", Types.STRING)

.field("number", Types.INT)

)

.inAppendMode()

.registerTableSource("source");

// 計算天級別的uv

// Table table &＃61; tEnv.sqlQuery("select DateUtil(rowtime),count(distinct fruit) from source group by DateUtil(rowtime)");

// 计算小时级别uv

Table table &＃61; tEnv.sqlQuery("select DateUtil(rowtime,&＃39;yyyyMMddHH&＃39;),count(distinct fruit) from source group by DateUtil(rowtime,&＃39;yyyyMMddHH&＃39;)");

tEnv.toRetractStream(table, Row.class).addSink(new SinkFunction>() {

&＃64;Override

public void invoke(Tuple2 value, Context context) throws Exception {

System.out.println(value.f1.toString());

}

});

System.out.println(env.getExecutionPlan());

env.execute("ComputeUVDay");

}

}

其中DateUtil类如下&＃xff1a;

package org.table.uv;

import org.apache.flink.table.functions.ScalarFunction;

import java.sql.Timestamp;

import java.text.DateFormat;

import java.text.SimpleDateFormat;

public class DateUtil extends ScalarFunction {

public static String eval(long timestamp){

String result &＃61; "null";

try {

DateFormat sdf &＃61; new SimpleDateFormat("yyyy-MM-dd");

result &＃61; sdf.format(new Timestamp(timestamp));

} catch (Exception e) {

e.printStackTrace();

}

return result;

}

public static String eval(long ts, String format) {

String result &＃61; "null";

try {

DateFormat sdf &＃61; new SimpleDateFormat(format);

result &＃61; sdf.format(ts);

} catch (Exception e) {

e.printStackTrace();

}

return result;

}

public static void main(String[] args) {

String eval &＃61; eval(System.currentTimeMillis(),"yyyyMMddHH");

System.out.println(eval);

}

}

代码里面的案例&＃xff0c;是可以用于生产中的吗&＃xff1f;

假如数据量小可以直接使用&＃xff0c;每秒数据量大的话&＃xff0c;就比较麻烦。因为你看group by后面的维度&＃xff0c;只有当天date 这个维度&＃xff0c;这样就会导致计算状态超级集中而使得内存占用超大进而引发oom。

这种情况解决办法就是将状态打散&＃xff0c;然后再次聚合即可&＃xff0c;典型的分治思想。

具体做法作为福利分享给球友吧。

还有一个问题就是由于存在全局去重及分组操作&＃xff0c;flink内部必然要维护一定的状态信息&＃xff0c;那么这些状态信息肯定不是要一直保存的&＃xff0c;比如uv&＃xff0c;我们只需要更新今天&＃xff0c;最多昨天的状态&＃xff0c;这个点之前的状态要删除的&＃xff0c;不能让他白白占着内存&＃xff0c;而导致任务内存消耗巨大&＃xff0c;甚至因oom而挂掉。

StreamQueryConfig streamQueryConfig &＃61; tEnv.queryConfig();

streamQueryConfig.withIdleStateRetentionTime(Time.minutes(10),Time.minutes(15));

tEnv.sqlUpdate(sql,streamQueryConfig);

再有就是能使用事件时间吗&＃xff1f;事件时间假如事件严重超时了&＃xff0c;比如&＃xff0c;我们状态保留时间设置的是两天&＃xff0c;两天之后状态清除&＃xff0c;那么这时候来了事件时间刚刚好是两天之前的&＃xff0c;由于已经没有状态就会重新计算uv覆盖已经生成的值&＃xff0c;就导致值错误了&＃xff0c;这个问题如何解决呢&＃xff1f;

这算是一个疑问吧&＃xff1f;