个性化举荐系统源码之之收集用户行径数据

m5k1umn

在举荐系统中另一个必不可少的数据便是用户行径数据，能够说用户行径数据是举荐系统的基石，巧妇难为无米之炊，因此接下来，咱们就要将用户的行径数据同步到举荐系统数据库中。

在文案举荐系统中，用户行径包含揭发、点击、停留、保藏、分享等，因此这儿咱们定义的用户行径数据的字段包含：出现时间（actionTime）、停留时间（readTime）、频道 ID（channelId）、事件名叫作（action）、用户 ID（userId）、文案 ID（articleId）以及算法 ID（algorithmCombine），这儿采用 json 格式，如下所示

# 揭发的参数 {"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:15:36","readTime":"","channelId":18,"param":{"action": "click", "userId": "2", "articleId": "18577", "algorithmCombine": "C2"}} {"actionTime":"2019-04-10 18:15:38","readTime":"1621","channelId":18,"param":{"action": "read", "userId": "2", "articleId": "18577", "algorithmCombine": "C2"}} {"actionTime":"2019-04-10 18:15:39","readTime":"","channelId":18,"param":{"action": "click", "userId": "1", "articleId": "14299", "algorithmCombine": "C2"}} {"actionTime":"2019-04-10 18:15:39","readTime":"","channelId":18,"param":{"action": "click", "userId": "2", "articleId": "14299", "algorithmCombine": "C2"}} {"actionTime":"2019-04-10 18:15:41","readTime":"914","channelId":18,"param":{"action": "read", "userId": "2", "articleId": "14299", "algorithmCombine": "C2"}} {"actionTime":"2019-04-10 18:15:47","readTime":"7256","channelId":18,"param":{"action": "read", "userId": "1", "articleId": "14299","algorithmCombine": "C2"}}

用户离线行径数据

因为用户行径数据规模庞大，一般是每日更新一次，以供离线计算运用。首要，在 Hive 中创建用户行径数据库 profile 及用户行径表 user_action，设置根据日期进行分区，并匹配 json 格式

-- 创建用户行径数据库 create database if not exists profile comment "use action" location /user/hive/warehouse/profile.db/; -- 创建用户行径信息表 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 # 匹配json格式 LOCATION /user/hive/warehouse/profile.db/user_action;

一般用户行径数据被保留在应用服务器的日志文件中，咱们能够利用 Flume 监听应用服务器上的日志文件，将用户行径数据收集到 Hive 的 user_action 表对应的 HDFS 目录中，Flume 配置如下

a1.sources= s1a1.sinks = k1 a1.channels = c1 a1.sources.s1.channels= c1 a1.sources.s1.type = exec a1.sources.s1.command = tail -F /root/logs/userClick.log a1.sources.s1.interceptors=i1 i2 a1.sources.s1.interceptors.i1.type=regex_filter a1.sources.s1.interceptors.i1.regex=\\{.*\\} a1.sources.s1.interceptors.i2.type=timestamp# c1 a1.channels.c1.type=memory a1.channels.c1.capacity=30000 a1.channels.c1.transactionCapacity=1000 # k1 a1.sinks.k1.type=hdfs a1.sinks.k1.channel=c1 a1.sinks.k1.hdfs.path=hdfs://192.168.19.137:9000/user/hive/warehouse/profile.db/user_action/%Y-%m-%d a1.sinks.k1.hdfs.useLocalTimeStamp = true a1.sinks.k1.hdfs.fileType=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 a1.sinks.k1.hdfs.idleTimeout=60

编写 Flume 起步脚本 collect_click.sh

#!/usr/bin/env bash export JAVA_HOME=/root/bigdata/jdk export HADOOP_HOME=/root/bigdata/hadoop export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin /root/bigdata/flume/bin/flume-ng agent -c /root/bigdata/flume/conf -f /root/bigdata/flume/conf/collect_click.conf -Dflume.root.logger=INFO,console -name a1

Flume 自动生成目录后，需要手动相关 Hive 分区后才可加载到数据

alter table user_action add partition (dt=2019-11-11) location "/user/hive/warehouse/profile.db/user_action/2011-11-11/"

用户实时行径数据

为了加强举荐的实时性，咱们亦需要收集用户的实时行径数据，以供在线计算运用。这儿利用 Flume 将日志收集到 Kafka，在线计算任务能够从 Kafka 读取用户实时行径数据。首要，开启 zookeeper，以保护进程运行

/root/bigdata/kafka/bin/zookeeper-server-start.sh -daemon /root/bigdata/kafka/config/zookeeper.properties

开启 Kafka

/root/bigdata/kafka/bin/kafka-server-start.sh /root/bigdata/kafka/config/server.properties # 开启信息生产者 /root/bigdata/kafka/bin/kafka-console-producer.sh --broker-list 192.168.19.19092 --sync --topic click-trace # 开启消费者 /root/bigdata/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.19.137:9092 --topic click-trace

修改 Flume 的日志收集配置文件，添加 c2 和 k2 ，将日志数据收集到 Kafka

a1.sources = s1 a1.sinks = k1 k2 a1.channels = c1 c2 a1.sources.s1.channels= c1 c2 a1.sources.s1.type = exec a1.sources.s1.command = tail -F /root/logs/userClick.log a1.sources.s1.interceptors=i1 i2 a1.sources.s1.interceptors.i1.type=regex_filter a1.sources.s1.interceptors.i1.regex=\\{.*\\} a1.sources.s1.interceptors.i2.type=timestamp # c1 a1.channels.c1.type=memory a1.channels.c1.capacity=30000 a1.channels.c1.transactionCapacity=1000 # c2 a1.channels.c2.type=memory a1.channels.c2.capacity=30000a1.channels.c2.transactionCapacity=1000 # k1 a1.sinks.k1.type=hdfs a1.sinks.k1.channel=c1 a1.sinks.k1.hdfs.path=hdfs://192.168.19.137:9000/user/hive/warehouse/profile.db/user_action/%Y-%m-%d a1.sinks.k1.hdfs.useLocalTimeStamp =truea1.sinks.k1.hdfs.fileType=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 a1.sinks.k1.hdfs.idleTimeout=60# k2 a1.sinks.k2.channel=c2 a1.sinks.k2.type=org.apache.flume.supervisorctl 咱们能够利用supervisorctl来管理supervisor。sink.kafka.KafkaSink a1.sinks.k2.kafka.bootstrap.servers=192.168.19.137:9092 a1.sinks.k2.kafka.topic=click-trace a1.sinks.k2.kafka.batchSize=20 a1.sinks.k2.kafka.producer.requiredAcks=1

编写 Kafka 起步脚本 start_kafka.sh

#!/usr/bin/env bash # 起步zookeeper/root/bigdata/kafka/bin/zookeeper-server-start.sh -daemon /root/bigdata/kafka/config/zookeeper.properties# 起步kafka /root/bigdata/kafka/bin/kafka-server-start.sh /root/bigdata/kafka/config/server.properties # 增多topic/root/bigdata/kafka/bin/kafka-topics.sh --zookeeper 192.168.19.137:2181 --create --replication-factor 1 --topic click-trace --partitions 1

进程管理

咱们这儿运用 Supervisor 进行进程管理，当进程反常时能够自动重启，Flume 进程配置如下

[program:collect-click] command=/bin/bash /root/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

Kafka 进程配置如下

[program:kafka] command=/bin/bash /root/toutiao_project/scripts/start_kafka.shuser=root autorestart=true redirect_stderr=true stdout_logfile=/root/logs/kafka.log loglevel=info stopsignal=KILL stopasgroup=true killasgroup=true

起步 Supervisor

supervisord -c /etc/supervisord.conf

起步 Kafka 消费者，并在应用服务器日志文件中写入日志数据，Kafka 消费者就可收集到实时行径数据

# 起步Kafka消费者/root/bigdata/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.19.137:9092 --topic click-trace# 写入日志数据 echo{\"actionTime\":\"2019-04-10 21:04:39\",\"readTime\":\"\",\"channelId\":18,\"param\":{\"action\": \"click\", \"userId\": \"2\", \"articleId\": \"14299\", \"algorithmCombine\": \"C2\"}} >> userClick.log# 消费者接收到日志数据 {"actionTime":"2019-04-10 21:04:39","readTime":"","channelId":18,"param":{"action": "click", "userId": "2", "articleId": "14299", "algorithmCombine": "C2"}}

Supervisor 常用命令如下

supervisorctl > status # 查看程序状态 > start apscheduler # 起步apscheduler单一程序 > stop toutiao:* # 关闭toutiao组程序 > start toutiao:* # 起步toutiao组程序 > restart toutiao:* # 重启toutiao组程序 > update # 重启配置文件修改过的程序

b1gc8v

回顾历史，我们不难发现：无数先辈用鲜血和生命铺就了中华民族复兴的康庄大道。

4lqedz

在遇到你之前，我对人世间是否有真正的圣人是怀疑的。

wrjc1hod

顶楼主，说得太好了！

7wu1wm0

i免费外链发布平台 http://www.fok120.com/