文章推薦系統(tǒng) | 三倒得、收集用戶行為數(shù)據(jù)

推薦閱讀：
文章推薦系統(tǒng) | 一泻红、推薦流程設(shè)計
文章推薦系統(tǒng) | 二、同步業(yè)務(wù)數(shù)據(jù)

在上一篇文章中霞掺，我們完成了業(yè)務(wù)數(shù)據(jù)的同步谊路，在推薦系統(tǒng)中另一個必不可少的數(shù)據(jù)就是用戶行為數(shù)據(jù)，可以說用戶行為數(shù)據(jù)是推薦系統(tǒng)的基石菩彬，巧婦難為無米之炊缠劝，所以接下來，我們就要將用戶的行為數(shù)據(jù)同步到推薦系統(tǒng)數(shù)據(jù)庫中骗灶。

在文章推薦系統(tǒng)中惨恭，用戶行為包括曝光、點(diǎn)擊耙旦、停留脱羡、收藏、分享等免都，所以這里我們定義的用戶行為數(shù)據(jù)的字段包括：發(fā)生時間（actionTime）锉罐、停留時間（readTime）、頻道 ID（channelId）绕娘、事件名稱（action）脓规、用戶 ID（userId）、文章 ID（articleId）以及算法 ID（algorithmCombine）险领，這里采用 json 格式侨舆，如下所示

# 曝光的參數(shù)
{"actionTime":"2019-04-10 18:15:35","readTime":"","channelId":0,"param":{"action": "exposure", "userId": "2", "articleId": "[18577, 14299]", "algorithmCombine": "C2"}}

# 對文章觸發(fā)行為的參數(shù)
{"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"}}

用戶離線行為數(shù)據(jù)

由于用戶行為數(shù)據(jù)規(guī)模龐大升酣，通常是每天更新一次，以供離線計算使用态罪。首先噩茄，在 Hive 中創(chuàng)建用戶行為數(shù)據(jù)庫 profile 及用戶行為表 user_action，設(shè)置按照日期進(jìn)行分區(qū)复颈，并匹配 json 格式

-- 創(chuàng)建用戶行為數(shù)據(jù)庫
create database if not exists profile comment "use action" location '/user/hive/warehouse/profile.db/';
-- 創(chuàng)建用戶行為信息表
create table user_action
(
    actionTime STRING comment "user actions time",
    readTime   STRING comment "user reading time",
    channelId  INT comment "article channel id",
    param MAP<STRING, STRING> comment "action parameter"
)
    COMMENT "user primitive action"
    PARTITIONED BY (dt STRING) # 按照日期分區(qū)
    ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe' # 匹配json格式
    LOCATION '/user/hive/warehouse/profile.db/user_action';

通常用戶行為數(shù)據(jù)被保存在應(yīng)用服務(wù)器的日志文件中绩聘，我們可以利用 Flume 監(jiān)聽?wèi)?yīng)用服務(wù)器上的日志文件，將用戶行為數(shù)據(jù)收集到 Hive 的 user_action 表對應(yīng)的 HDFS 目錄中耗啦，F(xiàn)lume 配置如下

a1.sources = s1
a1.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 自動生成目錄后凿菩，需要手動關(guān)聯(lián) Hive 分區(qū)后才能加載到數(shù)據(jù)

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

用戶實時行為數(shù)據(jù)

為了提高推薦的實時性，我們也需要收集用戶的實時行為數(shù)據(jù)帜讲，以供在線計算使用衅谷。這里利用 Flume 將日志收集到 Kafka，在線計算任務(wù)可以從 Kafka 讀取用戶實時行為數(shù)據(jù)似将。首先获黔，開啟 zookeeper，以守護(hù)進(jìn)程運(yùn)行

/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

# 開啟消息生產(chǎn)者
/root/bigdata/kafka/bin/kafka-console-producer.sh --broker-list 192.168.19.19092 --sync --topic click-trace
# 開啟消費(fèi)者
/root/bigdata/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.19.137:9092 --topic  click-trace

修改 Flume 的日志收集配置文件在验，添加 c2 和 k2 玷氏，將日志數(shù)據(jù)收集到 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=30000
a1.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 = 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

# 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

進(jìn)程管理

我們這里使用 Supervisor 進(jìn)行進(jìn)程管理腋舌，當(dāng)進(jìn)程異常時可以自動重啟盏触，F(xiàn)lume 進(jìn)程配置如下

[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 進(jìn)程配置如下

[program:kafka]
command=/bin/bash /root/toutiao_project/scripts/start_kafka.sh
user=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 消費(fèi)者，并在應(yīng)用服務(wù)器日志文件中寫入日志數(shù)據(jù)块饺，Kafka 消費(fèi)者即可收集到實時行為數(shù)據(jù)

# 啟動Kafka消費(fèi)者
/root/bigdata/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.19.137:9092 --topic  click-trace

# 寫入日志數(shù)據(jù)
echo {\"actionTime\":\"2019-04-10 21:04:39\",\"readTime\":\"\",\"channelId\":18,\"param\":{\"action\": \"click\", \"userId\": \"2\", \"articleId\": \"14299\", \"algorithmCombine\": \"C2\"}} >> userClick.log

# 消費(fèi)者接收到日志數(shù)據(jù)
{"actionTime":"2019-04-10 21:04:39","readTime":"","channelId":18,"param":{"action": "click", "userId": "2", "articleId": "14299", "algorithmCombine": "C2"}}

Supervisor 常用命令如下

supervisorctl

> status              # 查看程序狀態(tài)
> start apscheduler   # 啟動apscheduler單一程序
> stop toutiao:*      # 關(guān)閉toutiao組程序
> start toutiao:*     # 啟動toutiao組程序
> restart toutiao:*   # 重啟toutiao組程序
> update              # 重啟配置文件修改過的程序

參考

https://www.bilibili.com/video/av68356229
https://pan.baidu.com/s/1-uvGJ-mEskjhtaial0Xmgw（學(xué)習(xí)資源已保存至網(wǎng)盤赞辩， 提取碼：eakp）

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 人面猴
    序言：七十年代末，一起剝皮案震驚了整個濱河市授艰，隨后出現(xiàn)的幾起案子辨嗽，更是在濱河造成了極大的恐慌，老刑警劉巖想诅，帶你破解...
    沈念sama閱讀 206,311評論 6贊 481
  • 死咒
    序言：濱河連續(xù)發(fā)生了三起死亡事件召庞，死亡現(xiàn)場離奇詭異，居然都是意外死亡来破，警方通過查閱死者的電腦和手機(jī)篮灼，發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 88,339評論 2贊 382
  • 救了他兩次的神仙讓他今天三更去死
    文/潘曉璐 我一進(jìn)店門，熙熙樓的掌柜王于貴愁眉苦臉地迎上來徘禁，“玉大人诅诱，你說我怎么就攤上這事∷椭欤” “怎么了娘荡？”我有些...
    開封第一講書人閱讀 152,671評論 0贊 342
  • 道士緝兇錄：失蹤的賣姜人
    文/不壞的土叔 我叫張陵干旁，是天一觀的道長。 經(jīng)常有香客問我炮沐，道長争群，這世上最難降的妖魔是什么？ 我笑而不...
    開封第一講書人閱讀 55,252評論 1贊 279
  • ?港島之戀（遺憾婚禮）
    正文 為了忘掉前任大年，我火速辦了婚禮换薄，結(jié)果婚禮上，老公的妹妹穿的比我還像新娘翔试。我一直安慰自己轻要，他們只是感情好，可當(dāng)我...
    茶點(diǎn)故事閱讀 64,253評論 5贊 371
  • 惡毒庶女頂嫁案：這布局不是一般人想出來的
    文/花漫 我一把揭開白布垦缅。 她就那樣靜靜地躺著冲泥，像睡著了一般。 火紅的嫁衣襯著肌膚如雪壁涎。 梳的紋絲不亂的頭發(fā)上凡恍，一...
    開封第一講書人閱讀 49,031評論 1贊 285
  • 城市分裂傳說
    那天，我揣著相機(jī)與錄音粹庞，去河邊找鬼咳焚。 笑死，一個胖子當(dāng)著我的面吹牛庞溜，可吹牛的內(nèi)容都是我干的。 我是一名探鬼主播碑定，決...
    沈念sama閱讀 38,340評論 3贊 399
  • 雙鴛鴦連環(huán)套：你想象不到人心有多黑
    文/蒼蘭香墨 我猛地睜開眼流码，長吁一口氣：“原來是場噩夢啊……” “哼！你這毒婦竟也來了延刘？” 一聲冷哼從身側(cè)響起漫试，我...
    開封第一講書人閱讀 36,973評論 0贊 259
  • 萬榮殺人案實錄
    序言：老撾萬榮一對情侶失蹤，失蹤者是張志新（化名）和其女友劉穎碘赖，沒想到半個月后驾荣，有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體，經(jīng)...
    沈念sama閱讀 43,466評論 1贊 300
  • ?護(hù)林員之死
    正文 獨(dú)居荒郊野嶺守林人離奇死亡普泡，尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 35,937評論 2贊 323
  • ?白月光啟示錄
    正文 我和宋清朗相戀三年播掷，在試婚紗的時候發(fā)現(xiàn)自己被綠了。 大學(xué)時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片撼班。...
    茶點(diǎn)故事閱讀 38,039評論 1贊 333
  • 活死人
    序言：一個原本活蹦亂跳的男人離奇死亡歧匈，死狀恐怖，靈堂內(nèi)的尸體忽然破棺而出砰嘁，到底是詐尸還是另有隱情件炉，我是刑警寧澤勘究，帶...
    沈念sama閱讀 33,701評論 4贊 323
  • ?日本核電站爆炸內(nèi)幕
    正文 年R本政府宣布，位于F島的核電站斟冕，受9級特大地震影響口糕，放射性物質(zhì)發(fā)生泄漏。R本人自食惡果不足惜磕蛇，卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 39,254評論 3贊 307
  • 男人毒藥：我在死后第九天來索命
    文/蒙蒙 一景描、第九天 我趴在偏房一處隱蔽的房頂上張望。 院中可真熱鬧孤里，春花似錦伏伯、人聲如沸。這莊子的主人今日做“春日...
    開封第一講書人閱讀 30,259評論 0贊 19
  • 一樁弒父案说搅，背后竟有這般陰謀
    文/蒼蘭香墨 我抬頭看了看天上的太陽。三九已至虏等，卻和暖如春弄唧，著一層夾襖步出監(jiān)牢的瞬間，已是汗流浹背霍衫。 一陣腳步聲響...
    開封第一講書人閱讀 31,485評論 1贊 262
  • 情欲美人皮
    我被黑心中介騙來泰國打工候引， 沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留，地道東北人敦跌。 一個月前我還...
    沈念sama閱讀 45,497評論 2贊 354
  • 代替公主和親
    正文 我出身青樓澄干，卻偏偏與公主長得像，于是被迫代替她去往敵國和親柠傍。 傳聞我的和親對象是個殘疾皇子麸俘，可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 42,786評論 2贊 345