電商大數(shù)據(jù)項(xiàng)目（二）-推薦系統(tǒng)實(shí)戰(zhàn)之實(shí)時(shí)分析以及離線(xiàn)分析

電商大數(shù)據(jù)項(xiàng)目-推薦系統(tǒng)實(shí)戰(zhàn)（一）環(huán)境搭建以及日志，人口，商品分析
https://blog.51cto.com/6989066/2325073
電商大數(shù)據(jù)項(xiàng)目-推薦系統(tǒng)實(shí)戰(zhàn)之推薦算法
https://blog.51cto.com/6989066/2326209
電商大數(shù)據(jù)項(xiàng)目-推薦系統(tǒng)實(shí)戰(zhàn)之實(shí)時(shí)分析以及離線(xiàn)分析
https://blog.51cto.com/6989066/2326214

五、實(shí)時(shí)分析Top IP（實(shí)時(shí)分析Top用戶(hù)）
一）模塊介紹
電商網(wǎng)站運(yùn)營(yíng)中，需要分析網(wǎng)站訪(fǎng)問(wèn)排名前N的IP，主要用來(lái)審計(jì)是否有異常IP，同時(shí)對(duì)網(wǎng)站運(yùn)營(yíng)情況進(jìn)行分析。
（二）需求分析
①　如何統(tǒng)計(jì)IP
?通過(guò)用戶(hù)點(diǎn)擊日志，即可計(jì)算

②　如何分析IP
?分析不同時(shí)間段IP的訪(fǎng)問(wèn)量
?分析熱門(mén)IP

③　深度思考：如何去除爬蟲(chóng)水軍
（三）技術(shù)方案
?數(shù)據(jù)采集邏輯（ETL）
?用戶(hù)的訪(fǎng)問(wèn)日志（點(diǎn)擊日志）一般存儲(chǔ)在日志服務(wù)器，需要通過(guò)Flume拉取

?點(diǎn)擊日志的緩存
?由Flume采集的用戶(hù)點(diǎn)擊日志，緩存到Kafka中

?實(shí)時(shí)分析Top用戶(hù)信息
?使用Apache Storm進(jìn)行實(shí)時(shí)分析
?使用Spark Streaming進(jìn)行實(shí)時(shí)分析

?注意：在A(yíng)mbari部署的Flume和Kafka環(huán)境中，已經(jīng)將Flume和Kafka進(jìn)行了集成，所以可以直接將Flume采集的數(shù)據(jù)送入Kafka

（四）實(shí)驗(yàn)數(shù)據(jù)及說(shuō)明

?表user_click_log（用戶(hù)點(diǎn)擊信息表）
列名 描述 數(shù)據(jù)類(lèi)型 空/非空 約束條件
user_id 用戶(hù)ID varchar(18) Not null
user_ip 用戶(hù)IP varchar(20) Not null
url 用戶(hù)點(diǎn)擊URL varchar(200)
click_time 用戶(hù)點(diǎn)擊時(shí)間 varchar(40)
action_type 動(dòng)作名稱(chēng) varchar(40)
area_id 地區(qū)ID varchar(40)
補(bǔ)充說(shuō)明 action_type：1收藏，2加入購(gòu)物車(chē)，3點(diǎn)擊

?結(jié)果表：表hopip （熱門(mén)ip表）
列名 描述 數(shù)據(jù)類(lèi)型 空/非空 約束條件
ip IP varchar(18) Not null
pv 訪(fǎng)問(wèn)量 varchar(200)
補(bǔ)充說(shuō)明

（五）技術(shù)實(shí)現(xiàn)
①　通過(guò)Flume采集用戶(hù)點(diǎn)擊日志數(shù)據(jù)
?創(chuàng)建Flume的配置文件：

?注意：HDP集群kafka broker的默認(rèn)端口是6667，而不是9092
②　使用Kafka緩存數(shù)據(jù)
?新建一個(gè)名為mytopic的topic
bin/kafka-topics.sh --create --zookeeper hdp21:2181 --replication-factor 1 --partitions 1 --topic mytopic

?查看新創(chuàng)建的Topic
bin/kafka-topics.sh --list --zookeeper hdp21:2181

?測(cè)試：創(chuàng)建個(gè)消費(fèi)者來(lái)消費(fèi)mytopic里面的數(shù)據(jù)。bootstrap-server這里填主機(jī)名或者IP，而不是localhost
bin/kafka-console-consumer.sh --bootstrap-server hdp21:6667 --topic mytopic --from-beginning

?刪除Topic
bin/kafka-topics.sh --delete --zookeeper hdp21:2181 --topic mytopic
注意：這句命令只會(huì)將topic標(biāo)識(shí)為“刪除狀態(tài)”。如果想，徹底刪除Topic，需要將delete.topic.enable=true，并且重啟Kafka
③　實(shí)時(shí)分析Top 5用戶(hù)信息：基于Storm和Spark Streaming
?實(shí)現(xiàn)方式一：使用Apache Storm進(jìn)行實(shí)時(shí)分析

注意：
?pom文件中，Storm的版本為1.1.0
?在此pom文件中，已經(jīng)集成了Storm與Kafka、Redis、JDBC、MySQL的依賴(lài)

運(yùn)行結(jié)果，如下：

?實(shí)現(xiàn)方式二：使用Spark Streaming進(jìn)行實(shí)時(shí)分析
?結(jié)合Spark SQL分析Top用戶(hù)（熱點(diǎn)用戶(hù)）
?在上一章的Spark工程中的pom.xml文件中加入以下依賴(lài)

?注意：由于Kafka版本的問(wèn)題，接收Kafka的數(shù)據(jù)需要使用Receiver方式

運(yùn)行結(jié)果如下：

六、實(shí)時(shí)分析黑名單用戶(hù)
（一）模塊介紹
電商網(wǎng)站運(yùn)營(yíng)中，需要分析網(wǎng)站訪(fǎng)問(wèn)排名前N的客戶(hù)，主要用來(lái)審計(jì)是否有異常用戶(hù)，同時(shí)分析忠誠(chéng)用戶(hù)。
（二）需求分析
①　如何定義異常用戶(hù)？
?通過(guò)用戶(hù)訪(fǎng)問(wèn)頻度進(jìn)行分析，每小時(shí)訪(fǎng)問(wèn)40次以上，同時(shí)訪(fǎng)問(wèn)平均間隔4秒以?xún)?nèi)，則為異常用戶(hù)
?關(guān)聯(lián)異常用戶(hù)庫(kù)，進(jìn)行異常分析

②　如何計(jì)算用戶(hù)訪(fǎng)問(wèn)內(nèi)容
?通過(guò)用戶(hù)點(diǎn)擊日志，即可計(jì)算

③　深度思考：如何去除爬蟲(chóng)水軍
（三）技術(shù)方案
?數(shù)據(jù)采集邏輯（ETL）
?用戶(hù)的訪(fǎng)問(wèn)日志（點(diǎn)擊日志）一般存儲(chǔ)在日志服務(wù)器，需要通過(guò)Flume拉取

?點(diǎn)擊日志的緩存
?由Flume采集的用戶(hù)點(diǎn)擊日志，緩存到Kafka中

?實(shí)時(shí)分析Top用戶(hù)信息
?使用Apache Storm進(jìn)行實(shí)時(shí)分析黑名單用戶(hù)
?使用Spark Streaming進(jìn)行實(shí)時(shí)分析黑名單用戶(hù)

?注意：在A(yíng)mbari部署的Flume和Kafka環(huán)境中，已經(jīng)將Flume和Kafka進(jìn)行了集成，所以可以直接將Flume采集的數(shù)據(jù)送入Kafka

（四）實(shí)驗(yàn)數(shù)據(jù)及說(shuō)明

?表UserInfo（用戶(hù)信息表）
列名 描述 數(shù)據(jù)類(lèi)型 空/非空 約束條件
UserID ×××號(hào) varchar(18) Not null
Username 姓名 varchar(20) Not null
Sex 性別 varchar(10) Not null
Birthday 出生年月 datetime Not null
Birthprov 出生省份 varchar(8) Not null
Birthcity 出生城市 varchar(8) Not null
Job 工作 varchar(20) Not null
EducationLevel 教育水平 int Not null
SnnualSalary 年薪 double Not null
Addr_prov 現(xiàn)居地省份編號(hào) varchar(8) Not null
Addr_city 現(xiàn)居地城市編號(hào) varchar(8) Not null
Address 通信地址 varchar(50) Not null
Mobile 聯(lián)系電話(huà) varchar(11) Not null
Mail 郵箱 varchar(30) Not null
status 用戶(hù)狀態(tài) Int
補(bǔ)充說(shuō)明

?表user_click_log（用戶(hù)點(diǎn)擊信息表）
列名 描述 數(shù)據(jù)類(lèi)型 空/非空 約束條件
user_id 用戶(hù)ID varchar(18) Not null
user_ip 用戶(hù)IP varchar(20) Not null
url 用戶(hù)點(diǎn)擊URL varchar(200)
click_time 用戶(hù)點(diǎn)擊時(shí)間 varchar(40)
action_type 動(dòng)作名稱(chēng) varchar(40)
area_id 地區(qū)ID varchar(40)
補(bǔ)充說(shuō)明 action_type：1收藏，2加入購(gòu)物車(chē)，3點(diǎn)擊

?上一章的結(jié)果表：表hopip （熱門(mén)ip表）
列名 描述 數(shù)據(jù)類(lèi)型 空/非空 約束條件
user_ip IP varchar(18) Not null
pv 訪(fǎng)問(wèn)量 varchar(200)
補(bǔ)充說(shuō)明

?結(jié)果表：表black_list （黑名單表）
列名 描述 數(shù)據(jù)類(lèi)型 空/非空 約束條件
user_id 用戶(hù)ID varchar(18) Not null
user_ip 用戶(hù)IP varchar(40) Not null
補(bǔ)充說(shuō)明
（五）技術(shù)實(shí)現(xiàn)
1.首先，在第五章的需求里面，我們已經(jīng)實(shí)現(xiàn)了Hot IP的分析。下面是Hop IP的結(jié)果表。
?上一章的結(jié)果表：表hopip （熱門(mén)ip表）
列名 描述 數(shù)據(jù)類(lèi)型 空/非空 約束條件
user_ip IP varchar(18) Not null
pv 訪(fǎng)問(wèn)量 varchar(200)
補(bǔ)充說(shuō)明
2.我們只需要根據(jù)需求的規(guī)定，關(guān)聯(lián)用戶(hù)信息表，即可分析出某段時(shí)間內(nèi)的黑名單用戶(hù)信息，例如：每10秒鐘，統(tǒng)計(jì)過(guò)去30秒內(nèi)訪(fǎng)問(wèn)頻率超過(guò)10次的用戶(hù)信息。這時(shí)候就需要使用到窗口函數(shù)。在A(yíng)pache Storm和Spark Streaming中都可以使用窗口函數(shù)。

3.方案一：使用Storm的窗口函數(shù)，將結(jié)果寫(xiě)入MySQL
?在MySQL中創(chuàng)建數(shù)據(jù)庫(kù)和相應(yīng)的表：
create database demo;
CREATE USER 'demo'@'%'IDENTIFIED BY 'Welcome_1';
GRANT ALL PRIVILEGES ON . TO 'demo'@'%';
FLUSH PRIVILEGES;
create table myresult(userid int primary key,PV int);
?注意：
?在上一章的課程中，pom文件已經(jīng)集成了Storm與Kafka、Redis、JDBC、MySQL的依賴(lài)
?如果使用Storm的提供JdbcInsertBolt組件，會(huì)一直將結(jié)果不停寫(xiě)入MySQL中。更好的做法是：創(chuàng)建一個(gè)自己的Bolt組件：如果MySQL不存在該userid，就執(zhí)行插入；如果已經(jīng)存在就執(zhí)行更新操作。

當(dāng)分析出每個(gè)用戶(hù)的PV后，可在MySQL中執(zhí)行下面的查詢(xún)，查看黑名單用戶(hù)信息。
select userinfo.userid,userinfo.username,myresult.PV from userinfo,myresult where userinfo.userid=myresult.userid;
4.方案二：使用Spark Streaming的窗口函數(shù)

輸出結(jié)果：

七、廣告點(diǎn)擊流量統(tǒng)計(jì)
（一）模塊介紹
電商網(wǎng)站運(yùn)營(yíng)中，廣告是一個(gè)非常重要的模塊，需要分析廣告的點(diǎn)擊情況，主要用來(lái)優(yōu)化每個(gè)城市的點(diǎn)擊量。

（二）需求分析
①　如何分析廣告點(diǎn)擊數(shù)據(jù)？
?通過(guò)用戶(hù)對(duì)廣告的點(diǎn)擊log，分析廣告數(shù)據(jù)

②　計(jì)算每天各省各城市各廣告的點(diǎn)擊量

（三）技術(shù)方案
離線(xiàn)分析：拉取廣告日志，需要通過(guò)Flume拉取到HDFS上，通過(guò)MapReduce和Spark進(jìn)行離線(xiàn)分析。
（四）實(shí)驗(yàn)數(shù)據(jù)及說(shuō)明

?廣告點(diǎn)擊日志
列名 描述 數(shù)據(jù)類(lèi)型 空/非空 約束條件
userid 用戶(hù)id varchar(18) Not nul
ip 點(diǎn)擊IP varchar(18) Not null
click_time 點(diǎn)擊時(shí)間 varchar(20) Not null
url 廣告鏈接 varchar(20)
area_id 地區(qū)ID varchar(20)

?表Area_info（地區(qū)信息表）
列名 描述 數(shù)據(jù)類(lèi)型 空/非空 約束條件
area_id 地區(qū)編號(hào) varchar(18) Not null
area_name 地區(qū)名稱(chēng) varchar(20) Not null

（五）技術(shù)實(shí)現(xiàn)
①　使用Flume采集用戶(hù)點(diǎn)擊日志
?通常使用shell腳本執(zhí)行日志采集
?復(fù)雜情況，使用可視化的ETL工具，來(lái)進(jìn)行Flume Agent控制
下面是配置文件，注意HDFS的端口號(hào)。

②　使用Hive進(jìn)行離線(xiàn)廣告分析
?創(chuàng)建地區(qū)表
create external table areainfo
(areaid int,areaname string)
row format delimited fields terminated by ','
location '/input/project07';

?創(chuàng)建廣告點(diǎn)擊日志表
create external table adloginfo
(userid int,ip string,clicktime string,url string,areaid int)
row format delimited fields terminated by ','
location '/flume/20180603';

?通過(guò)SQL分析數(shù)據(jù)
select areainfo.areaname,adloginfo.url,adloginfo.clicktime,count(adloginfo.clicktime)
from adloginfo,areainfo
where adloginfo.areaid=areainfo.areaid
group by areainfo.areaname,adloginfo.url,adloginfo.clicktime;
③　使用Spark進(jìn)行離線(xiàn)廣告日志分析

④　使用Pig進(jìn)行離線(xiàn)廣告日志分析
?加載地區(qū)表
areainfo = load '/input/areainfo.txt' using PigStorage(',') as (areaid:int,areaname:chararray);

?加載廣告日志表
adloginfo = load '/flume/20180603/userclicklog.txt' using PigStorage(',') as (userid:int,ip:chararray,clicktime:chararray,url:chararray,areaid:int);

?按照url、地區(qū)、點(diǎn)擊時(shí)間第廣告點(diǎn)擊日志進(jìn)行分組
adloginfo1 = group adloginfo by (url,areaid,clicktime);

?提取url、areaid、點(diǎn)擊時(shí)間和總頻率
adloginfo2 = foreach adloginfo1 generate group,COUNT(adloginfo.clicktime);

?執(zhí)行多表查詢(xún)，關(guān)聯(lián)地區(qū)表
result = join areainfo by areaid, adloginfo2 by group.areaid;