1. 写在前面

这里是有关于一个头条推荐项目的学习笔记，主要是整理工业上的推荐系统用到的一些常用技术， 这是第二篇，第一篇介绍的是数据库的迁移技术，也就是mysql中的数据如何导入到Hadoop中，在大数据平台上进行处理， 而这一篇要介绍的技术是用户的行为日志收集，通过flume将业务数据服务器A的日志收集到hadoop服务器的hdfs的hive中。 对于头条文章推荐来说， 用户的行为(点击，收藏，转发等)代表着每一次的喜好反馈，能够了解分析用户的行为，喜好变化，也能为后面为建立用户画像提供依据，所以用户行为数据对于我们搞推荐非常重要， 那么用户的点击行为我们如何收集起来并保存到Hive*后面分析使用呢？ 就是这篇文章要介绍的主要技术了。主要包括：

• 用户行为日志收集的相关工作流程
• 如何进行flume，hive等收集的相关配置
• 应用supervisior管理flume并进行实时收集点击日志

Ok, let’s go!

2. 用户行为日志收集的相关工作流程

2.1 埋点开发测试流程

埋点，就是将用户的行为记录埋点到日志当中。一般用户有很多日志， 这里的推荐场景会把用户的点击行为统一到行为日志，其实还有其他业务场景如(下单日志，支付日志)

• 埋点参数(哪些行为是需要我们记录的)

  • 就是在应用特定的流程中收集一些信息，用来跟踪应用使用的情况，后续用来进一步优化产品或是提供运营的数据支撑
  • 重要性：埋点数据是推荐系统的基石，模型训练和效果数据统计都基于埋点数据，需要保证埋点数据的正确无误

• 流程：

  1. PM(项目经理)、算法推荐工程师一起指定埋点需求文档
  2. 后端、客户端APP集成
  3. 推荐人员基于文档埋点测试与数理

我们的任务就是第一点，埋点需求整理出来，剩下的问题交给其他开发人员了。

2.2 黑马头条文章推荐埋点需求整理

用户对于文章，一般会产生点击，浏览时间，收藏，分享等行为， 我们基于这些来制定我们的需求：

• 埋点场景： 首页的各个频道推荐

• 埋点事件号：停留时间read， 点击事件click， 曝光事件exposure(相当于刷新一次请求推荐新文章), 收藏事件collect, 分享事件share, 这样就确定好了埋点的字段参数

• 埋点参数文件结构(埋到日志里面的一个文件结构):
  我们将埋点参数设计成一个固定格式的json字符串，它包含了事件发生事件、算法推荐号、获取行为的频道号、帖子id列表、帖子id、用户id、事件号字段。

  # 曝光的参数， 曝光时间，阅读时间，哪个频道，参数{行为，用户，推荐文章，推荐算法}
  {"actionTime":"2019-04-10 18:15:35","readTime":"","channelId":0,"param":{"action": "exposure", "userId": "2", "articleId": "[18577, 14299]", "algorithmCombine": "C2"}}
  
  # 对文章发生行为的参数 点击时间，阅读时间毫秒，频道，参数(行为名称，哪个用户，哪篇文章，哪个算法推荐的)
  {"actionTime":"2019-04-10 18:12:11","readTime":"2886","channelId":18,"param":{"action": "read", "userId": "2", "articleId": "18005", "algorithmCombine": "C2"}}
  {"actionTime":"2019-04-10 18:15:32","readTime":"","channelId":18,"param":{"action": "click", "userId": "2", "articleId": "18005", "algorithmCombine": "C2"}}
  {"actionTime":"2019-04-10 18:15:34","readTime":"1053","channelId":18,"param":{"action": "read", "userId": "2", "articleId": "18005", "algorithmCombine": "C2"}}
  {"actionTime":"2019-04-10 18:15:36","readTime":"","channelId":18,"param":{"action": "click", "userId": "2", "articleId": "18577", "algorithmCombine": "C2"}}
  

  曝光的参数： 用户下拉刷新行为，推荐新的若干篇文章，我们要记录下给他新推荐的文章id和给他推荐的算法。根据后面用户的真实点击情况，就能统计出用户的点击比例，有利于我们做调整。

好了， 理论介绍完毕，下面又是我自己摸索的时候了，因为一开始我就不是按照这个课程走的，包括各种环境，数据其实都不和课程匹配，所以没有办法跟着人家老师的敲一遍，只能听完原理之后，在自己的环境中摸索，不足就是会遇到很多坑，浪费一些时间，优势就是对知识更能加深理解和体会，两面性吧哈哈，下面开始干。

3. 离线部分 - 用户的日志收集

先说下要做的事情： 通过flume，将业务数据服务器A的用户点击行为日志收集到hadoop服务器hdfs的hive中。

然后说下我目前的情况， 和上一篇类似， 我这里拿到的数据其实是人家从flume收集好了存储到了hdfs上，我需要做的仅仅在Hive中建立表然后管理上这些数据即可。 但是我不太满足哈哈，这根本学不到啥东西，其实我对前面的这一步其实是更好奇的， 所以我想自己摸索第一步，就是如何从flume收集日志数据，并存到hdfs上，但是我没有数据，也没有手把手指导，并不知道能不能摸索出来，管他呢， 先干再说， 走一步看一步吧， 因为总是想的话总感觉全是问题，且根本找不到切入点，所以不如先干，有些东西动起手来之后，反而会清晰不少。

先打开三台虚拟机，开启hadoop环境再说(具体搭建过程我已经写了7篇文章来记录，可以去大数据学习笔记里面看)。 那么我怎么玩呢？ 我也没有个头绪其实， 看到课程里面说的收集步骤，分为下面三步：

1. 创建Hive对应的日志收集表(收集到新的数据库中)
2. flume收集日志配置
3. 开启收集命令

3.1 创建Hive对应的日志收集表

第一步这个是这样子的，首先要在Hive中创建一个新的数据库，叫做profile.db， 后面会在里面建立一个表user_action，把收集到的用户行为存到这个表下。 因为我看他给的数据库和文件目录就是这样的哈哈：


这些数据如果直接用的话非常简单，和上一篇一样，导入HDFS，然后再Hive里面建表关联即可。但我很好奇flume是如何收集用户数据的，所以下面我打算，先自己造点日志，走一遍flume收集用户日志的流程，然后再回来按照他的数据建表。所以我造点日志，就拿上面的行为日志就好了，在master上，创建一个logs目录，然后在里面创建一个userClick.log文件，在里面写入下面的内容(行为时间给他改改，我看人家用了分区的操作，按照天把行为放在了一块):

这就相当于用户的点击日志行为了哈哈。下面就是看看如何通过flume把这个日志读到hdfs上去。

3.2 flume读取日志文件到hdfs

3.2.1 flume简单介绍

这里说先先介绍下flume， 由于我也是现学，这里直接参考下面某个大佬整理的一点内容了，链接放到了最后面。

• Flume 是分布式的海量日志采集、聚合和传输的系统。Flume 基于流式架构（指一条一条采集数据），灵活简单。

Flume最主要的作用就是，实时读取服务器本地磁盘的数据，将数据写入到HDFS。说白了就是一个采集数据的工具，使用简单，配置文件即可。

可以从文件，文件夹，http协议等地方采集（目录：只要目录中的文件有变化，就会进行采集该数据, Shell命令：执行一条shell命令，shell命令的输出会进sources）

Flume的优点：

1. flume可以和任意存储进程集成
2. 数据的数据速率大于写入目的存储的速率，flume会进行缓冲，减小hdfs的压力（怎么理解：源数据来的忽快忽慢，日志信息可能在用户活跃期间瞬间有5G日志信息，在夜间等几乎没有日志信息，但是因为flume的缓冲能力，到达hdfs的速度相对平稳一些。）

Flume的运行机制：lume运行的最小单元，独立运行在一个JVM中。一个agent里面包括一个或多个sources,channels,sinks，每个agent内部有三个组件。

1. souce: 数据采集组件，对接源数据
2. channel: 传输通道组件，通俗叫管道，数据的缓冲区，连接source和sink，将source和sink进行打通
3. sink: 下沉组件，用于向下一级agent传递数据或者往最终的存储系统传递数据。

好了，关于理论知识就介绍到这里，主要是看咋用， 关于flume的安装和搭建，可以看我这篇文章, 在这里面还写了个flume如何监听某个端口数据打印到控制台， 而这里我们更进一步，是把本地文件采集到HDFS上。

3.2.2 读取设置

首先，得需要给flume一个配置文件，告诉flume如何读取， 又放到哪里去。 进入flume/conf目录，创建一个collect_click.conf文件，写入flume的配置。指定好是三个组件内容：

• sources: 为实时查看文件末尾，interceptors解析的json文件
• channels: 指定内存存储，并且制定batchData的大小，PutList和TakeList的大小见参数，Channel总容量大小见参数
• sink: 形式直接到hdfs，以及路径，文件大小策略默认1024、event数量策略、文件闲置时间

把下面这个代码写进去：

# 定义这个agent中各个组件的名字， 这里的agent取名a1, 三个组件取名s1,k1,c1
a1.sources = s1     # 定义source
a1.sinks = k1       # 定义 sink 
a1.channels = c1    # 定义 channel

# 描述和配置source channel sink之间的连接关系
a1.sources.s1.channels= c1
a1.sinks.k1.channel=c1    

# 描述和配置source组件r1，注意不能往监控目录中丢重复同名文件呢， 从哪里取数据
a1.sources.s1.type = exec   # 定义source类型为执行文件， 定义为目录用spooldir
a1.sources.s1.command = tail -F /home/icss/workspace/toutiao_project/logs/userClick.log   # 本地log文件所在目录
a1.sources.s1.interceptors=i1 i2
a1.sources.s1.interceptors.i1.type=regex_filter
a1.sources.s1.interceptors.i1.regex=\\{.*\\}  # json数据字符串的解析格式
a1.sources.s1.interceptors.i2.type=timestamp

# #描述和配置channel组件：c1
a1.channels.c1.type=memory
a1.channels.c1.capacity=30000   # 这里的数根据实际业务场景变，每天实时数据量大的时候，这里会相应的增大
a1.channels.c1.transactionCapacity=1000

# # 描述和配置sink组件k1   存储到哪？
a1.sinks.k1.type=hdfs     # sink类型为hdfs
a1.sinks.k1.hdfs.path=hdfs://192.168.56.101:9000/user/hive/warehouse/profile.db/user_action/%Y-%m-%d      # 这个就是指明存储的路径  后面是指定分区的时候，用时间来命名下一层目录
a1.sinks.k1.hdfs.useLocalTimeStamp = true  # 是否使用本地时间戳
a1.sinks.k1.hdfs.fileType=DataStream  # 生成的文件类型，默认是Sequencefile，可以用DataStream（就是普通文本）
a1.sinks.k1.hdfs.writeFormat=Text  
a1.sinks.k1.hdfs.rollInterval=0    # 多久生成新文件，单位是多少秒
a1.sinks.k1.hdfs.rollSize=10240    # 多大生成新文件  也就是如果传过来的数据大小不足这个，就不会创建新文件
a1.sinks.k1.hdfs.rollCount=0    #多少个event生成新文件
a1.sinks.k1.hdfs.idleTimeout=60  # 文件占用时间，这个也挺重要的

# 文件占用时间这个，表示如果过了这个时间，就自动关闭文件，关闭flume占用
# 假设文件1， 写了1000个行为之后，flume依然一直占用着这个文件的话，我们是无法读取到这1000个行为的
# 所以为了满足实时的分析， 我们一般会设置这个文件占用时间，也就是如果flume占用了60分钟还没有任何操作的话，就关闭掉这个占用

这个直接按照它这个走， 在flume目录下创建job文件夹并进入job文件， 在里面创建一个collect_click.conf文件，写入flume的配置即可。

flume这边算是配置完成， 下面配置下hive。

3.2.3 Hive设置

接下来， 就是设置Hive了， 因为上面的flume的设置，其实就完成了flume的收集工作， 只要运行上面的配置文件，flume就能够将本地的用户行为日志上传到HDFS的profile.db的user_action表里面。 接下来，就是在Hive中创建数据库，然后创建user_action表与profile.db里面的user_action表进行关联。

打开Hive， 创建profile数据库，表示用户相关数据画像存储到这个数据库中。

create database if not exists profile comment "use action" location '/user/hive/warehouse/profile.db/';

在profile数据库中创建user_action表，并指定格式， 这里要注意格式，和之前的不太一样了：

create table user_action(
actionTime STRING comment "user actions time",
readTime STRING comment "user reading time",
channelId INT comment "article channel id",
param map comment "action parameter")
COMMENT "user primitive action"
PARTITIONED BY(dt STRING)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
LOCATION '/user/hive/warehouse/profile.db/user_action';

在Hive中输入这段代码，迎来了第一个报错FAILED: ParseException line 5:10 mismatched input 'comment' expecting < near 'map' in map type， 原因是map集合需要指定键值类型， 第5行按个改成这样即可：

param map<string, string> comment "action parameter")

再次建立表，迎来了第二个报错：FAILED: Execution Error, return code 1 from org.apache.hadoop.hive.ql.exec.DDLTask. Cannot validate serde: org.apache.hive.hcatalog.data.JsonSerDe， 果然，不是和人家一样的环境就是不行啊，步步有坑。 这个的解决方式： 需要将支持序列化的hive-hcatalog-core-2.1.1.jar添加进来,执行如下命令即可解决该问题， 这个在安装目录里面找。我的是

所以输入下面的命令：

hive> add jar /opt/bigdata/hive/hive2.1/hcatalog/share/hcatalog/hive-hcatalog-core-2.1.1.jar;

这样，就建立好了用户行为表了。关于上面那段代码说几个细节了：

1. ROW FORMAT SERDE ‘org.apache.hive.hcatalog.data.JsonSerDe’: 添加一个json格式匹配
2. LOCATION ‘/user/hive/warehouse/profile.db/user_action’： 与这个位置的表进行关联
3. PARTITIONED BY(dt STRING)： 这个很重要，这个说的是Hive中的分区，首先要知道Hive为啥要分区？
   • 随着表的不断增大，对于新纪录的增加，查找，删除等(DML)的维护也更加困难。对于数据库中的超大型表，可以通过把它的数据分成若干个小表，从而简化数据库的管理活动，对于每一个简化后的小表，我们称为一个单个的分区。
   • hive中分区表实际就是对应hdfs文件系统上独立的文件夹，该文件夹内的文件是该分区所有数据文件。
   • 分区可以理解为分类，通过分类把不同类型的数据放到不同的目录下。分类的标准就是分区字段，可以一个，也可以多个。这里的dt，就是指的分区字段
   • 分区表的意义在于优化查询。查询时尽量利用分区字段。如果不使用分区字段，就会全部扫描。这里是根据点击行为时间进行的分区（日期)
     
     

Hive这边设置完成。下面就可以开启flume收集命令了。

3.2.4 开启收集命令

cd /opt/bigdata/flume/flume1.9

bin/flume-ng agent --conf conf/ --conf-file job/collect_click.conf --name a1 -Dflume.root.logger=INFO,console

这样就搜集完了。我们看下效果：


这里面注意最下面这个红框上面的.tmp文件，这时候说明flume正在占用文件，此时我们是无法用hive查看这个文件里面的数据的， 等flume占用完之后，生成了下面的FlumeData文件。这时候我们就能打开或者是用hive查看了。打开看下：

原来这个日期是指的系统时间的日期啊， 今天正好3月9号，所以这里把我自己造的用户点击行为全都搜集到了3月9号里面去了。下面回到hive， 我们尝试看下数据：

发现竟然没有数据。这是因为Hive表的分区没有与flume收集到的目录关联， 没有关联是查询不到数据。所以这里还需要手动关联分区下：

# 如果flume自动生成目录后，需要手动关联分区
alter table user_action add partition (dt='2021-03-09') location "/user/hive/warehouse/profile.db/user_action/2021-03-09/"

见证奇迹的时刻到了：

这样，用户的点击行为日志就存到了Hive表中啦。下面我们再尝试写入向userClick.log一条新的用户行为日志， 为了标识， 我这里直接用今天的时间

echo {\"actionTime\":\"2021-03-09 21:04:39\",\"readTime\":\"\",\"channelId\":18,\"param\":{\"action\": \"click\", \"userId\": \"2\", \"articleId\": \"14299\", \"algorithmCombine\": \"C2\"}} >> userClick.log

# 再写一条：
echo {\"actionTime\":\"2021-03-09 22:00:00\",\"readTime\":\"\",\"channelId\":18,\"param\":{\"action\": \"click\", \"userId\": \"2\", \"articleId\": \"14299\", \"algorithmCombine\": \"C2\"}} >> userClick.log

结果如下：

再看hdfs：

看时间也能知道已经传到hdfs上了，这时候在从hive中查询：

这样，就把前面那个步骤，从本地的用户行为日志通过flume，实时传到HDFS上，并用hive进行查询的过程就探索完啦，还好一切顺利。 这里再把注意的点整理下：

1. 创建Hive对应的日志收集表这里
   ①Hive数据库的设置过程， 分区这块很重要
   ②手动关联分区的hadoop目录才能查询
2. flume收集日志配置
   这里的重点就是给flume配置文件，在里面告诉flume从哪里收集数据到哪里就OK
3. 有Hive的user_action表， 并且hadoop有相应目录， flume会自动生成目录，但如果想要通过spark SQL获取内容的话，每天每次还是要主动关联的，后面会提及，如果每天每次手动关联，那实在是太累了。

好了， 探索结束， 下面就可以删除掉这些用他给的用户行为数据在Hive中建表了哈哈。

下面还有个东西要整理，就是自动化运行，和上一篇一样，我们的这个日志收集行为是天天都要运行的，所以就需要一个工具来自动化的管理，这个叫做Supervisor， 这是个进程管理工具。

4. 应用supervisior管理flume并进行实时收集点击日志

4.1 supervisior进程管理

Supervisor作为进程管理工具，很方便的监听、启动、停止、重启一个或多个进程。用Supervisor管理的进程，当一个进程意外被杀死，supervisort监听到进程死后，会自动将它重新拉起，很方便的做到进程自动恢复的功能，不再需要自己写shell脚本来控制。

首先先安装， 这里尝试用yum直接安装，提示没找到包：

解决方法， 下面的命令走一遍，注意切换成root：

# 安装Linux的epel的yum源的命令，某些yum源会提示无supervisor源码包，此时可以使用此命令
yum install epel-release

# 再安装
yum install -y supervisor

# 设置开机自启
systemctl enable supervisord

# 启动supervisord服务
systemctl start supervisord
# 重启
systemctl restart supervisord

# 查看supervisord服务状态
ps -ef|grep supervisord 

名称说明：

• supervisor：要安装的软件的名称。
• supervisord：装好supervisor软件后，supervisord用于启动supervisor服务。
• supervisorctl：用于管理supervisor配置文件中program。

在启动spervisord服务的时候， 又开始报错：

这时候根据提示输入后面那个命令查看，说：pkg_resources.DistributionNotFound: The 'supervisor==3.4.0' distribution was not found and is required by the application， 这个的原因是supervisor对python3支持不好，须使用python2， 也就是Python版本引起的， 编辑/usr/bin/supervisord文件，将首行的#!/usr/bin/python修改为#!/usr/bin/python2即可。这样再启动就起来了。

下面进行配置：

运行echo_supervisord_conf命令输出默认的配置项，可以如下操作将默认配置保存到文件中

echo_supervisord_conf > supervisord.conf

这时候又报

这个和上面同样的处理方式，找到这个文件，然后把首行的python改成python2即可。这时候会在当前目录下生成一个supervisord.conf的配置文件，打开它，修改：

include选项指明包含的其他配置文件。这里是配置supervisor开启的配置文件在哪里。

好吧，我的不是这样操作，我打开上面那个配置文件之后啥也没有，所以百度了一下，我的在/etc/有个supervisord.conf配置文件了，我需要在那里面改。看来不同的版本都不一样啊。

把这个改了就行了files=supervisor/*.conf。

然后我们在/etc目录下新建子目录supervisor（与配置文件里的选项相同），并在/etc/supervisor/中新建头条推荐管理的配置文件reco.conf。后面有啥关于supervisor都可以往这里面加。

加入配置模板如下(模板)：

# 这里是举了两个supervisor自动管理进程的两个例子，只需要关注我注释的这3行，其他默认配置就OK
[program:recogrpc]   # 管理进程的名字
command=/root/anaconda3/envs/reco_sys/bin/python /root/headlines_project/recommend_system/ABTest/routing.py
directory=/root/headlines_project/recommend_system/ABTest   # 指定执行路径的一个命令
user=root       
autorestart=true
redirect_stderr=true
stdout_logfile=/root/logs/reco.log  # 管理过程中信息报错的打印路径
loglevel=info
stopsignal=KILL
stopasgroup=true
killasgroup=true

[program:kafka]
command=/bin/bash /root/headlines_project/scripts/startKafka.sh
directory=/root/headlines_project/scripts
user=root
autorestart=true
redirect_stderr=true
stdout_logfile=/root/logs/kafka.log
loglevel=info
stopsignal=KILL
stopasgroup=true
killasgroup=true

待会我们会根据这个模板去配置我们这里的flume自动收集日志的进程， 下面先说配置完成之后的事情。

配置完成之后，我们就开启supervisor了。

supervisord -c /etc/supervisord.conf

#查看supervisor是否运行
ps aux | grep supervisord

我们还可以利用supervisorctl来管理supervisor。

supervisorctl

> status    # 查看程序状态
> start apscheduler  # 启动 apscheduler 单一程序  这个名词就是我们上面模板中program后面的那个程序名词
> stop toutiao:*   # 关闭 toutiao组 程序
> start toutiao:*  # 启动 toutiao组 程序
> restart toutiao:*    # 重启 toutiao组 程序
> update    # 重启配置文件修改过的程序  一旦修改配置文件，就需要执行这个

这里执行这个，又是会报python版本的错误，所以先改这东西。

vim /usr/bin/supervisorctl
# 将首行python改成python2

现在我status下，啥也没有，因为我还没有设置要管理的程序。

工具介绍完了，下面就是我们这次的目标了，启动监听flume收集日志程序。

4.2 启动监听flume收集日志程序

目的： 启动监听flume收集日志

• 我们将启动flume的程序建立成collect_click.sh脚本

我把脚本统一写到了我的头条项目里面的scripts里面，在这里面建立collect_click.sh脚本，然后输入：

#!/usr/bin/env bash

export JAVA_HOME=/opt/bigdata/java/jdk1.8
export HADOOP_HOME=/opt/bigdata/hadoop/hadoop2.8
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin

/opt/bigdata/flume/flume1.9/bin/flume-ng agent --conf /opt/bigdata/flume/flume1.9/conf/ --conf-file /opt/bigdata/flume/flume1.9/job/collect_click.conf --name a1 -Dflume.root.logger=INFO,console

这里需要指定JAVA_HOME和HADOOP_HOME，因为supervisor和终端是分开的，否则找不到。并且这里都要换成绝对路径了。

接下来，就是配置superior了，让它能够自动的运行上面这个脚本，类似一个守护进程在那里跑着。编辑上面建立的reco.conf文件，

cd /etc/supervisor
vim reco.conf

在里面加入：

[program:collect-click]
command=/bin/bash /home/icss/workspace/toutiao_project/scripts/collect_click.sh
user=root
autorestart=true
redirect_stderr=true
stdout_logfile=/root/logs/collect.log
loglevel=info
stopsignal=KILL
stopasgroup=true
killasgroup=true

这时候，再去supervisor的控制台里面update一下。

这里显示加了一个程序了。就是我们上面的collect-click。再status下：

那老师说默认不会启动，我这怎么启动了呢？不管他了。这时候，打开日志查看，会发现flume在后台运行了：cat /root/logs/collect.log

再测试下，也就是再往之前的log日志里面添加一条用户数据，看看啥效果：

也就是flume在后台自动收取用户行为数据，并放到Hive中了。 好了，探索完毕，关掉这个程序了。

stop collect-click

到这里，把用户点击行为的自动收集过程探索完毕啦