一、说明

在实际应用中，我们往往会关注，到底有多少不同的用户访问了网站，所以另外一个统计流量的重要指标是网站的独立访客数（Unique Visitor，UV）。

二、数据准备

package com.lyh.flink06;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;@Data
@NoArgsConstructor
@AllArgsConstructor
public class UserBehavior {private Long userId;private Long itemId;private Integer categoryId;private String behavior;private Long timestamp;
}

三、思路

对于UserBehavior数据源来说，我们直接可以根据userId来区分不同的用户。
将userid放到SET集合里面，统计集合长度，便可以统计到网站的访客数。

四、代码

package com.lyh.flink06;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;import javax.naming.ldap.HasControls;
import java.util.HashSet;
import java.util.Set;public class PUcount {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);env.readTextFile("input/UserBehavior.csv").map(line -> {String[] data = line.split(",");return new UserBehavior(Long.valueOf(data[0]),Long.valueOf(data[1]),Integer.valueOf(data[2]),data[3],Long.valueOf(data[4]));}).filter(ub ->"pv".equals(ub.getBehavior())).keyBy(UserBehavior::getBehavior).process(new KeyedProcessFunction<String, UserBehavior, String>() {Set<Long> userIdset = new HashSet<>();@Overridepublic void processElement(UserBehavior value,Context ctx,Collector<String> out) throws Exception {userIdset.add(value.getUserId());out.collect("uv:"+ userIdset.size());}}).print();env.execute();}
}