离线数仓（二）【用户行为日志采集平台搭建】

用户行为日志采集平台搭建

1、用户行为日志概述

        用户行为日志的内容，主要包括用户的各项行为信息以及行为所处的环境信息。收集这些信息的主要目的是优化产品和为各项分析统计指标提供数据支撑。收集这些信息的手段通常为埋点。

        目前主流的埋点方式，有代码埋点（前端/后端）、可视化埋点、全埋点等。

  • 代码埋点是通过调用埋点SDK函数（JavaScript 技术 ），在需要埋点的业务逻辑功能位置调用接口，上报埋点数据。例如，我们对页面中的某个按钮埋点后，当这个按钮被点击时，可以在这个按钮对应的 OnClick 函数里面调用SDK提供的数据发送接口，来发送数据（把用户的行为以 http请求的形式发送到后端）。
  • 可视化埋点只需要研发人员集成采集 SDK，不需要写埋点代码，业务人员就可以通过访问分析平台的“圈选”功能，来“圈”出需要对用户行为进行捕捉的控件，并对该事件进行命名。圈选完毕后，这些配置会同步到各个用户的终端上，由采集 SDK 按照圈选的配置自动进行用户行为数据的采集和发送。
  • 全埋点是通过在产品中嵌入SDK（比如使用神策的埋点SDK），前端自动采集页面上的全部用户行为事件，上报埋点数据，相当于做了一个统一的埋点。然后再通过界面配置哪些数据需要在系统里面进行分析。

目前用的比较多的是全埋点，但是毕竟使用第三方的 SDK 可能出现数据泄露的风险。

2、用户行为日志内容

本项目收集和分析的用户行为信息主要有页面浏览记录、动作记录、曝光记录、启动记录和错误记录。

2.1、页面浏览记录

        页面浏览记录，记录的是访客对页面的浏览行为，该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息及页面信息等。

  • 用户 id：每个账号分配一个唯一的 id
  • 时间信息：用户跳入页面的时间和跳出的时间，以此来评估用户对该商品的感兴趣程度等等指标
  • 设备信息：获取用户的设备信息，可以用来做广告推荐等

2.2、动作记录

        动作记录，记录的是用户的业务操作行为，该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息 及动作目标对象信息等。

  • 动作目标对象相关信息，比如用户点击了哪个模块 （优惠券？哪个优惠价）

2.3、曝光记录

        曝光记录，记录的是曝光行为，该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息及曝光对象信息等。

  • 曝光相关信息：曝光指的是在当前界面展示过的内容

2.4、启动记录

启动记录，记录的是用户启动应用的行为，该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息、启动类型及开屏广告信息等。

  • 启动类型：用户是点击软件按钮启动还是推送通知启动
  • 开屏广告信息：打开软件时广告界面的广告id

2.5、错误记录

        启动记录，记录的是用户在使用应用过程中的报错行为，该行为的环境信息主要有用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息、以及可能与报错相关的页面信息、动作信息、曝光信息和动作信息。

3、用户行为日志格式

        上面我们知道了各种用户行为日志的类型，我们可以发现，有很多记录其实是重复的，比如用户信息、设别信息、渠道信息 ... 存储重复的记录不仅没有意义而且占用磁盘空间，那么应该怎样存储呢？

        我们的日志结构大致可分为两类，一是页面日志，二是启动日志。

3.1、页面日志

        页面日志，以页面浏览为单位，即一个页面浏览记录，生成一条页面埋点日志。一条完整的页面日志包含，一个页面浏览记录，若干个用户在该页面所做的动作记录，若干个该页面的曝光记录，以及一个在该页面发生的报错记录。除上述行为信息，页面日志还包含了这些行为所处的各种环境信息，包括用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息等。

{"common": {                     -- 环境信息"ar": "230000",             -- 地区编码"ba": "iPhone",             -- 手机品牌"ch": "Appstore",           -- 渠道"is_new": "1",              -- 是否首日使用，首次使用的当日，该字段值为1，过了24:00，该字段置为0。"md": "iPhone 8",           -- 手机型号"mid": "YXfhjAYH6As2z9Iq",  -- 设备id"os": "iOS 13.2.9",         -- 操作系统"uid": "485",               -- 会员id"vc": "v2.1.134"            -- app版本号},"actions": [{                   -- 动作(事件)"action_id": "favor_add",   -- 动作id"item": "3",                -- 目标id"item_type": "sku_id",      -- 目标类型"ts": 1585744376605         -- 动作时间戳}],"displays": [{                  -- 曝光"displayType": "query", -- 曝光类型"item": "3",            -- 曝光对象id"item_type": "sku_id",  -- 曝光对象类型"order": 1,             -- 出现顺序"pos_id": 2             -- 曝光位置},{"displayType": "promotion","item": "6","item_type": "sku_id","order": 2,"pos_id": 1},{"displayType": "promotion","item": "9","item_type": "sku_id","order": 3,"pos_id": 3},{"displayType": "recommend","item": "6","item_type": "sku_id","order": 4,"pos_id": 2},{"displayType": "query ","item": "6","item_type": "sku_id","order": 5,"pos_id": 1}],"page": {                          -- 页面信息"during_time": 7648,           -- 持续时间毫秒"item": "3", 	               -- 目标id"item_type": "sku_id",         -- 目标类型"last_page_id": "login",       -- 上页类型"page_id": "good_detail",      -- 页面ID"sourceType": "promotion"      -- 来源类型},                                 "err": {                           --错误"error_code": "1234",          --错误码"msg": "***********"           --错误信息},                                 "ts": 1585744374423                --跳入时间戳
}

3.2、启动日志

        启动日志以启动为单位，及一次启动行为，生成一条启动日志。一条完整的启动日志包括一个启动记录，一个本次启动时的报错记录，以及启动时所处的环境信息，包括用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息等。

{"common": {"ar": "370000","ba": "Honor","ch": "wandoujia","is_new": "1","md": "Honor 20s","mid": "eQF5boERMJFOujcp","os": "Android 11.0","uid": "76","vc": "v2.1.134"},"start": {   "entry": "icon",         --icon手机图标  notice 通知   install 安装后启动"loading_time": 18803,  --启动加载时间"open_ad_id": 7,        --广告页ID"open_ad_ms": 3449,    -- 广告总共播放时间"open_ad_skip_ms": 1989   --  用户跳过广告时点},
"err":{                     --错误
"error_code": "1234",      --错误码"msg": "***********"       --错误信息
},"ts": 1585744304000    --日志产生的时间
}

4、模拟数据

模拟数据我们直接使用封装好的日志生成 jar 包：

java -jar gmall2020-mock-log-2021-10-10.jar

 我们需要在 hadoop102 和 hadoop103 上都配置好日志数据源，所以为了方便还需要写一个脚本：

#!/bin/bash
for i in hadoop102 hadoop103; doecho "========== $i =========="ssh $i "cd /opt/module/applog/; java -jar gmall2020-mock-log-2021-10-10.jar >/dev/null 2>&1 &"
done

        我们在 hadoop102 和 hadoop103 的 /opt/module/applog/log 看到最新生成的日志，这个日志文件是按照时间滚动的

 用户行为日志数据采集模块

1、数据通道

        现在，我们需要把用户行为日志数据通过 Flume 传输到 Kafka 的日志 topic 中。

        上面我们的 hadoop102 和 hadoop103 已经具备了生成日志的能力，之前我们说 flume 是但节点安装，确实没错，但是这里因为我们是两个地方产生日志，如果跨节点传输日志那样效率太低了，所以索性就直接安装两个 flume 即可。

这里我们回顾一下之前学习 flume 的时候讲到的 source ，有

  • taildir：适合用于监听多个实时追加的文件（支持不同目录的不同文件），并且能够实现断点续传
  • netcat：监听 netcat 端口发生过来的内容
  • exec：适用于监控一个实时追加的文件，不能实现断点续传
  • spooling：适合用于同步新文件，支持断点续传，但是它监听的是文件夹而不是文件，不适合对实时追加日志的文件进行监听并同步
  • avro：当需要多个 flume 节点之间传输日志时，之间必须使用 avro sourcce 和 avro sink
  • kafka：kafka source 其实就是一个 Kafka 消费者，从Kafka 的 topic 中读取数据：

还有 channel：

  • file ：日志数据缓存在本地磁盘，并且在内存中保存有索引，磁盘中同样保存两份索引来保证读取的速度
  • memory：日志缓存在内存中，读取速度快，但是不安全
  • kafka：实际使用最多的

kafka channel 其实就是把 Event 保存到了 Kafka 集群当中，它支持 3 种结构：

1. 结合 source 和 sink 

2. 只结合 source 

 3. 只结合 sink

还有 sink：

  • hdfs：写出到 hdfs
  • kafka：写出到 kafka，相当于 kafka 消费者
  • avro：写出到 avro 系统（avro 是一个 rpc 数据序列化系统）

为了保证数据的可靠性（以及 Kafak 具有削峰解耦的功能），我们当然是选择 Kafak Channel 作为 channel ，那么对于 Kafak Channel 的三种结构，我们应该如何选择呢？

首先第一种结构 - 结合 source 和 sink，这种结构不可取。因为 Kafka Channel 有一个参数 parseAsFlumeEvent（默认为 true） ，它的意思是把日志数据以 Flume Event  的格式存储到 Kafka，这就包括了 Event Header 和 Event Body。对于离线数仓而言，因为它本来 sink 用的就是 flume 的 sink 从 kafka channel 中读取，它可以把 event 中的 body 解析出来。但是对于实时数仓，flink 是从 kafka 中直接读取数据的，所以它读到的就是包含 header 和 body 的完整 event 数据，而 header 对于实时数仓来说根本没有，所以没有必要存储。

既然 header 不需要存储我们能不能把 parseAsFlumeEvent 设为 false 然后继续使用这种结构呢，我们知道拦截器是在 source 和 channel 之间起作用的，如果不使用 flume event 的形式存储数据就不能使用拦截器，而 channel 和 sink 之间又不能设置拦截器。所以最好的办法就是使用第二种结构 - Kafak Channel 结合 Source 使用，同时设置 parseAsFlumeEvent 参数为 false ，这样存储进 Kafka 的数据就只有 body，虽然没有了 sink，但是我们可以把存储在 Kafak 中的数据当做一个数据源，下游再用一个 Source 读取，这样我们就可以再设置拦截器了。

2、作业脚本配置

编写 flume 配置文件

file_to_kafka.conf

#定义组件
a1.sources = r1
a1.channels = c1#配置source
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /opt/module/applog/log/app.*
a1.sources.r1.positionFile = /opt/module/flume-1.9.0/taildir_position.json#配置channel
a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092
a1.channels.c1.kafka.topic = topic_log
a1.channels.c1.parseAsFlumeEvent = false#组装 
a1.sources.r1.channels = c1

测试

# 启动 zookeeper 和 kafka
zk start
kf.sh start
# 开启flume作业
bin/flume-ng agent -n a1 -c conf/ -f job/warehouse/file_to_kafka.conf -Dflume.root.logger=INFO,console
# 开启kafka消费者
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_log
# 开启日志生成器
mklog.sh

运行结果：

至此，flume 的 source -> kafka channel 配置完毕，接下来我们需要在 flume 这里进行一个简单的 ETL 数据清洗，把一些脏数据去除掉，而且这里不能做复杂的拦截器，比如添加多个拦截器，因为flume 毕竟只是一个传递数据的管道，如果这里的 ETL工作太复杂会导致数据堆积在拦截器这里，影响效率。

添加拦截器

引入依赖和打包插件：

    <dependencies><dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-core</artifactId><version>1.9.0</version><scope>provided</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.62</version></dependency></dependencies><build><plugins><plugin><artifactId>maven-compiler-plugin</artifactId><version>2.3.2</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin><plugin><artifactId>maven-assembly-plugin</artifactId><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins></build>

Json 工具类：

package com.lyh.gmall.utils;import com.alibaba.fastjson.JSONObject;public class JSONUtil {// 通过异常捕获来校验json是否合法public static boolean isJsonValidate(String body) {try {JSONObject.parseObject(body);return true;}catch (Exception e){return false;}}
}

自定义拦截器：

package com.lyh.gmall.interceptor;import com.lyh.gmall.utils.JSONUtil;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;import java.nio.charset.StandardCharsets;
import java.util.List;public class ETLInterceptor implements Interceptor {@Overridepublic void initialize() {}// 单个事件的拦截器@Overridepublic Event intercept(Event event) {// 1. 获取body当中的数据String body = new String(event.getBody(), StandardCharsets.UTF_8);// 2. 判断数据是否合法,如果合法直接返回,否则返回 nullreturn JSONUtil.isJsonValidate(body)?event:null;}// 多个事件的拦截器@Overridepublic List<Event> intercept(List<Event> list) {for (int i = 0; i < list.size(); i++) {if (intercept(list.get(i)) == null){list.remove(i);i--;}}return list;}@Overridepublic void close() {}public static class Builder implements Interceptor.Builder {@Overridepublic Interceptor build() {return new ETLInterceptor();}@Overridepublic void configure(Context context) {}}
}

打包后会生成两个jar包，选择带依赖的放到 flume 的 lib 目录下，在 flume 作业配置文件中添加以下配置来添加拦截器：

a1.sources.r1.interceptors =  i1
a1.sources.r1.interceptors.i1.type = com.lyh.gmall.interceptor.ETLInterceptor$Builder

测试：

因为 flume 监听的是 applog/log/app-log-xxx.log 文件，所以我们直接 echo 写入来模拟错误日志（不完整的 json 信息）： 

可以看到，不合法的 json 数据被过滤掉了，至此，我们的数据采集通道可用了。根据我们的项目架构图可以看到，我们需要在 hadoop102 和 hadoop103 上采集，所以 hadoop103 也许要配置：

flume 采集脚本

#!/bin/bashcase $1 in
"start"){for i in hadoop102 hadoop103doecho " -------- $i flume开始采集 -------"ssh $i "nohup /opt/module/flume-1.9.0/bin/flume-ng agent -n a1 -c /opt/module/flume-1.9.0/conf/ -f /opt/module/flume-1.9.0/job/warehouse/file_to_kafka.conf >/dev/null 2>&1 &"done
};; 
"stop"){for i in hadoop102 hadoop103doecho " -------- $i flume停止采集-------"ssh $i "ps -ef | grep file_to_kafka | grep -v grep |awk  '{print \$2}' | xargs -n1 kill -9 "done
};;
esac

总结

        至此，用户行为日志的采集环境搭建好了，只要输入命令 f1.sh start 就可以让在 hadoop102 和 hadoop103 上的 flume开启采集日志了。接下来就是业务数据的采集了。

        这一部分还算简单，对于 flume 的操作无非就是拦截器的编写，逻辑也没多复杂，flume 的 source 使用了 taildir 监听由脚本不断生成的日志文件，channel 使用了 Kafka source ，sink 估计不会采用 flume 的 sink。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/475745.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【Kubernetes in Action笔记】1.快速开始

【Kubernetes in Action笔记】1.快速开始

在Kubernetes上运行一个程序 基础运行环境 当前的运行环境为使用虚拟机构建的单master集群。 [rootk8s-master ~]# kubectl get nodes NAME STATUS ROLES AGE VERSION k8s-master Ready control-plane 109d v1.27.1 k8s-node1 Ready …
阅读更多...
代码随想录算法训练营第十六天 | 104.二叉树的最大深度，111.二叉树的最小深度，222.完全二叉树的节点个数 [二叉树篇]

代码随想录算法训练营第十六天 | 104.二叉树的最大深度，111.二叉树的最小深度，222.完全二叉树的节点个数 [二叉树篇]

代码随想录算法训练营第十六天 LeetCode 104.二叉树的最大深度题目描述思路递归后序遍历参考代码递归前序遍历参考代码总结 LeetCode 111.二叉树的最小深度题目描述思路参考代码 LeetCode 222.完全二叉树的节点个数题目描述思路后序遍历参考代码1后序遍历参考代码2 LeetCode 10…
阅读更多...
Stable Diffusion ComfyUI安装详细教程

Stable Diffusion ComfyUI安装详细教程

上一篇文章介绍了sd-webui的安装教程&#xff0c;但学习一下ComfyUI这种节点流程式的对理解AI绘画有较大帮助&#xff0c;而且后期排查错误会更加方便&#xff0c;熟练后用这种方式做AI绘画可玩性会更多。 文章目录 一、安装包说明二、安装文件介绍三、安装步骤四、汉化五、云主…
阅读更多...
Linux-ls命令

Linux-ls命令

目录 ls&#xff1a;查看目录下文件/文件夹 ls -l&#xff1a;列表显示文件 ls -a&#xff1a;显示所有文件正常情况下‘ . ’开头的文件是隐藏的 ls -la&#xff1a;以列表形式显示所有文件包括隐藏文件 ls -lt&#xff1a;按时间倒序查看文件 ls -R&#xff1a;递归方式…
阅读更多...
Open CASCADE学习|曲线向曲面投影

Open CASCADE学习|曲线向曲面投影

在三维空间中&#xff0c;将曲线向曲面投影通常涉及复杂的几何计算。这个过程可以通过多种方法实现&#xff0c;但最常见的是使用数学和几何库&#xff0c;如OpenCASCADE&#xff0c;来处理这些计算。 在OpenCASCADE中&#xff0c;投影曲线到曲面通常涉及以下步骤&#xff1a;…
阅读更多...
第14讲投票帖子详情实现

第14讲投票帖子详情实现

投票帖子详情实现 后端,根据id查询投票帖子信息&#xff1a; /*** 根据id查询* param id* return*/ GetMapping("/{id}") public R findById(PathVariable(value "id")Integer id){Vote vote voteService.getById(id);WxUserInfo wxUserInfo wxUserInf…
阅读更多...
简单介绍数据结构的基本概念

简单介绍数据结构的基本概念

数据结构的基本概念 常用术语 数据 数据&#xff08;Data&#xff09;是客观事物的符号表示&#xff0c;是所有能输入到计算机中并被计算机程序处理的符号的总称。例如&#xff1a;整数、字符串、图形、图像、声音和动画等 数据元素 数据元素&#xff08;Data Element&…
阅读更多...
Shiro反弹shell和权限绕过含工具包

Shiro反弹shell和权限绕过含工具包

★★免责声明★★ 文章中涉及的程序(方法)可能带有攻击性&#xff0c;仅供安全研究与学习之用&#xff0c;读者将信息做其他用途&#xff0c;由Ta承担全部法律及连带责任&#xff0c;文章作者不承担任何法律及连带责任。 1、前言 反序列化漏洞原理和Shiro反序列化漏洞原理请参…
阅读更多...
安装 Windows 7

安装 Windows 7

1.镜像安装 镜像安装:安装Windows 7 2.安装过程(直接以图的形式呈现) 等待安装成功即可
阅读更多...
Linux——网络通信TCP通信常用的接口和tcp服务demo

Linux——网络通信TCP通信常用的接口和tcp服务demo

文章目录 TCP通信所需要的套接字socket()bind()listen()acceptconnect() 封装TCP socket TCP通信所需要的套接字 socket() socket()函数主要作用是返回一个描述符&#xff0c;他的作用就是打开一个网络通讯端口&#xff0c;返回的这个描述符其实就可以理解为一个文件描述符&a…
阅读更多...
防御-day5-防火墙综合实验

防御-day5-防火墙综合实验

一、实验拓扑 二、实验要求 1、办公区设备可以通过电信链路和移动链路上网(多对多的NAT、并且需要保留一个公网IP不能用来转换)。 2、分公司的设备可以通过总公司的移动链路和电信链路访问到DMZ区的http服务器 3、分公司内部的客户端可以通过公网地址访问到内部的服务器。 三、…
阅读更多...
0206-1-网络层

0206-1-网络层

第 4 章 网络层 网络层提供的两种服务 虚电路服务 数据报服务 概要: 虚电路服务与数据报服务的对比 网际协议 IP 网际协议 IP 是 TCP/IP 体系中两个最主要的协议之一。与 IP 协议配套使用的还有四个协议&#xff1a; 地址解析协议 ARP (Address Resolution Protocol)逆地…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023