另一方面，需求分析和构思UV，全名是Unique Visitor，即独立访问者，在实时计算中也被称为Dau(dailyactiveuser )，是每日活跃的用户。 为什么这么说呢，因为实时计算中的UV通常是指当天的访问者数量。
那么，从用户的行为日志中识别当天的访问者有两点。
第一，认识到该访问者打开的第一个页面，表示该访问者开始进入我们的APP应用程序
其二，访问者一天可以访问多次APP，所以我们可以在一天的范围内访问去重
1.1、核心过滤码先用keyby按mid分组，每组表示当前设备接入情况
//TODO 3.对读取的数据进行结构化的singleoutputstreamoperatorjsonobjectjsonobjds=jsonstrds.map (JSON str-JSON.parse object (JSON stringkeybywithmidds=jsonobjds.keyby (JSON obj-JSON obj.getjsonobject ) commonobject分组使用keystate状态，记录用户的进入时间
//TODO 5.过滤得到的uvsingleoutputstreamoperatorjsonobjectfilteredds=keybywithmidds.filter (newrichfilterfunctionjsonobjection 。 ) open方法的改写请参阅RichFilterFunction
//定义状态valuestatestringlastvisitdatestate=null//日期工具类定义simple date format SDF=null @ overridepublicvoidopen (configuration parameters ) throwsexception(/日期初始化工具类SDF=null //初始化状态valuestatedescriptorstringlastvisitdatestatedes=newvaluestatedescriptor (' lastvisitdatestate '，String.class )； //因为我们在计数日活DAU，所以状态数据只在那一天有效，过一天statettlconfigstatettlconfig=statettlconfig.new builder (time.days this.lastvisitdatestate=getruntimecontext ().getstate(lastvisitdatestatedes )； }重写filter方法以进行过滤
只要有上一页，就可以直接排除last_page_id为非空的字段，以表明这不是此用户第一次访问的页面。
状态用于记录用户的访问时间。 如果此lastVisitDate是今天，则用户今天访问了它，因此将其排除。 如果是空的或者不是今天，表示今天还没有访问，保留。
因为状态值主要用于筛选今天是否来，所以这个记录过了今天也几乎没有用。 这里enableTimeToLive设置了一天的过期时间，以避免状态过大。
@ overridepublicbooleanfilter (jsonobjectjsonobj ) Throwsexception首先，判断当前页面是否是从其他页面传来的stringlastpageid=JSON=nulllastpageid.length(0) { return false； //获取当前访问时间longts=JSONobj.getlong('ts )； //将当前访问时间戳转换为日期字符串string logdate=SDF.format (new date ) ts ) )； //状态日期stringlastvisitdate=lastvisitdatestate.value (； //当前页面的访问时间和状态时间中的if(lastvisitdate！=nulllastvisitdate.length (0last visit date.equals ) logdate ) ) { System.out.println (已访问： last visit date-' last ) } else { System.out.println ('未访问： lastVisitDate-' lastVisitDate '，||)； lastvisitdatestate.update (logdate； 返回真； } } 1.2、要将过滤后的UV写回Kafka的dwm_unique_visit //TODO 6.请将json写回string//6.1 JSON-stringsingleoutputstreamoooom 6.2kafka的dwm层Kafka ds.addsink (mykafkautil.getkafkasink ) sinktopic )写回1.3、测试开始logger.sh
在Idea上运行BaseLogApp
在Idea上运行UniqueVisitApp
显示控制台输出和kafka的dwm_unique_visit主题
运行流程
模拟生成数据-日志处理服务器-kafka的ODS层(ods_base_log )-BaseLogApp卸载-dwd_page_log-UniqueVisitApp读取，处理-