数据分析高级教程(二)

6 模块开发——ETL

该项目的数据分析过程在hadoop集群上实现，主要应用hive数据仓库工具，因此，采集并经过预处理后的数据，需要加载到hive数据仓库中，以进行后续的挖掘分析。

6.1创建原始数据表

--在hive仓库中建贴源数据表

drop table if exists ods_weblog_origin;

create table ods_weblog_origin(

valid string,

remote_addr string,

remote_user string,

time_local string,

request string,

status string,

body_bytes_sent string,

http_referer string,

http_user_agent string)

partitioned by (datestr  string)

row format delimited

fields terminated by  '\001';

 

点击流模型pageviews表

drop table if exists ods_click_pageviews;

create table ods_click_pageviews(

Session string,

remote_addr string,

time_local string,

request string,

visit_step string,

page_staylong string,

http_referer string,

http_user_agent string,

body_bytes_sent string,

status string)

partitioned by (datestr string)

row format delimited

fields terminated by '\001';

 

时间维表创建

drop table dim_time if exists ods_click_pageviews;

create table dim_time(

year string,

month string,

day string,

hour string)

row format delimited

fields terminated by ',';

 

 

6.2导入数据

导入清洗结果数据到贴源数据表ods_weblog_origin

load data inpath  '/weblog/preprocessed/16-02-24-16/' overwrite into table ods_weblog_origin partition(datestr='2013-09-18');

 

0:  jdbc:hive2://localhost:10000> show partitions ods_weblog_origin;

+-------------------+--+

|     partition     |

+-------------------+--+

| timestr=20151203  |

+-------------------+--+

 

0:  jdbc:hive2://localhost:10000> select count(*) from ods_origin_weblog;

+--------+--+

|  _c0    |

+--------+--+

| 11347  |

+--------+--+

 

导入点击流模型pageviews数据到ods_click_pageviews表

0:  jdbc:hive2://hdp-node-01:10000> load data inpath '/weblog/clickstream/pageviews'  overwrite into table ods_click_pageviews partition(datestr='2013-09-18');

 

0:  jdbc:hive2://hdp-node-01:10000> select count(1) from ods_click_pageviews;

+------+--+

| _c0  |

+------+--+

| 66   |

+------+--+

 

 

导入点击流模型visit数据到ods_click_visit表

 

 

 

6.3 生成ODS层明细宽表

6.3.1 需求概述

整个数据分析的过程是按照数据仓库的层次分层进行的，总体来说，是从ODS原始数据中整理出一些中间表（比如，为后续分析方便，将原始数据中的时间、url等非结构化数据作结构化抽取，将各种字段信息进行细化，形成明细表），然后再在中间表的基础之上统计出各种指标数据

 

6.3.2 ETL实现

建表——明细表   (源：ods_weblog_origin)   （目标：ods_weblog_detail）

drop table ods_weblog_detail;

create table  ods_weblog_detail(

valid           string, --有效标识

remote_addr     string, --来源IP

remote_user     string, --用户标识

time_local      string, --访问完整时间

daystr          string, --访问日期

timestr         string, --访问时间

month           string, --访问月

day             string, --访问日

hour            string, --访问时

request         string, --请求的url

status          string, --响应码

body_bytes_sent string,  --传输字节数

http_referer    string, --来源url[dht1] 

ref_host        string, --来源的host

ref_path        string, --来源的路径

ref_query       string, --来源参数query

ref_query_id    string, --来源参数query的值

http_user_agent string --客户终端标识

)

partitioned by(datestr  string);

 

--抽取refer_url到中间表  "t_ods_tmp_referurl"

--将来访url分离出host  path  query query id

drop table if exists  t_ods_tmp_referurl;

create table t_ ods _tmp_referurl  as

SELECT a.*,b.*

FROM ods_origin_weblog a  LATERAL VIEW parse_url_tuple(regexp_replace(http_referer,  "\"", ""), 'HOST', 'PATH','QUERY', 'QUERY:id') b as  host, path, query, query_id;

 

--抽取转换time_local字段到中间表明细表”t_ ods _detail”

drop table if exists t_ods_tmp_detail;

create table t_ods_tmp_detail  as

select  b.*,substring(time_local,0,10) as daystr,

substring(time_local,11)  as tmstr,

substring(time_local,5,2)  as month,

substring(time_local,8,2)  as day,

substring(time_local,11,2)  as hour

From t_ ods _tmp_referurl  b;

 

以上语句可以改写成：

insert into table  zs.ods_weblog_detail partition(datestr='$day_01')

select  c.valid,c.remote_addr,c.remote_user,c.time_local,

substring(c.time_local,0,10) as daystr,

substring(c.time_local,12)  as tmstr,

substring(c.time_local,6,2)  as month,

substring(c.time_local,9,2)  as day,

substring(c.time_local,11,3)  as hour,

c.request,c.status,c.body_bytes_sent,c.http_referer,c.ref_host,c.ref_path,c.ref_query,c.ref_query_id,c.http_user_agent

from

(SELECT

a.valid,a.remote_addr,a.remote_user,a.time_local,

a.request,a.status,a.body_bytes_sent,a.http_referer,a.http_user_agent,b.ref_host,b.ref_path,b.ref_query,b.ref_query_id  

FROM zs.ods_weblog_origin  a LATERAL VIEW parse_url_tuple(regexp_replace(http_referer,  "\"", ""), 'HOST', 'PATH','QUERY', 'QUERY:id') b as  ref_host, ref_path, ref_query, ref_query_id) c

"

0:  jdbc:hive2://localhost:10000> show partitions ods_weblog_detail;

+---------------------+--+

|      partition      |

+---------------------+--+

| dd=18%2FSep%2F2013  |

+---------------------+--+

1 row selected (0.134  seconds)

 

---

 

http://www.baidu.com/aapath?sousuoci=’angel’

 

parse_url_tuple(url,’HOST’,’PATH’,’QUERY’,’QUERY:id’)

 

 

 

7 模块开发——统计分析

注：每一种统计指标都可以跟各维度表进行叉乘，从而得出各个维度的统计结果 

篇幅限制，叉乘的代码及注释信息详见项目工程代码文件                   

为了在前端展示时速度更快，每一个指标都事先算出各维度结果存入mysql    

 

提前准备好维表数据，在hive仓库中创建相应维表，如：

时间维表：

create table v_time(year string,month string,day  string,hour string)

row format delimited

fields terminated by ',';

 

load data local inpath '/home/hadoop/v_time.txt' into  table v_time;

 

在实际生产中，究竟需要哪些统计指标通常由相关数据需求部门人员提出，而且会不断有新的统计需求产生，以下为网站流量分析中的一些典型指标示例。

1.    PV统计

1.1 多维度统计PV总量

1. 时间维度

--计算指定的某个小时pvs

select  count(*),month,day,hour from dw_click.ods_weblog_detail group by  month,day,hour;

 

 

--计算该处理批次（一天）中的各小时pvs

drop table dw_pvs_hour;

create table  dw_pvs_hour(month string,day string,hour string,pvs bigint) partitioned  by(datestr string);

 

insert into table  dw_pvs_hour partition(datestr='2016-03-18')

select a.month as  month,a.day as day,a.hour as hour,count(1) as pvs from ods_weblog_detail a

where  a.datestr='2016-03-18'  group by a.month,a.day,a.hour;

 

或者用时间维表关联

 

 

 

维度：日

drop table dw_pvs_day;

create table  dw_pvs_day(pvs bigint,month string,day string);

 

insert into table  dw_pvs_day

select count(1) as  pvs,a.month as month,a.day as day  from  dim_time a

join ods_weblog_detail b

on b.dd='18/Sep/2013' and  a.month=b.month and a.day=b.day

group by a.month,a.day;

 

--或者，从之前算好的小时结果中统计

Insert into table  dw_pvs_day

Select sum(pvs) as  pvs,month,day from dw_pvs_hour group by month,day having day='18';

 

结果如下：

 

维度：月

drop table t_display_pv_month;

create table t_display_pv_month (pvs bigint,month string);

insert into table t_display_pv_month

select count(*) as pvs,a.month from t_dim_time a

join  t_ods_detail_prt b on a.month=b.month group by a.month;

 

 

2. 按终端维度统计pv总量

注：探索数据中的终端类型

select  distinct(http_user_agent) from ods_weblog_detail where http_user_agent like  '%Mozilla%' limit 200;

 

 

终端维度：uc

drop table  t_display_pv_terminal_uc;

create table  t_display_pv_ terminal_uc (pvs bigint,mm string,dd string,hh string);

 

终端维度：chrome

drop table  t_display_pv_terminal_chrome;

create table t_display_pv_  terminal_ chrome (pvs bigint,mm string,dd string,hh string);

 

终端维度：safari

drop table  t_display_pv_terminal_safari;

create table  t_display_pv_ terminal_ safari (pvs bigint,mm string,dd string,hh string);

 

3. 按栏目维度统计pv总量

栏目维度：job

栏目维度：news

栏目维度：bargin

栏目维度：lane

 

 

1.2 人均浏览页数

需求描述：比如，今日所有来访者，平均请求的页面数

 

--总页面请求数/去重总人数

drop table  dw_avgpv_user_d;

create table  dw_avgpv_user_d(

day string,

avgpv string);

 

insert into table  dw_avgpv_user_d

select  '2013-09-18',sum(b.pvs)/count(b.remote_addr) from

(select remote_addr,count(1)  as pvs from ods_weblog_detail where datestr='2013-09-18' group by  remote_addr) b;

 

 

1.3 按referer维度统计pv总量

需求：按照来源及时间维度统计PVS，并按照PV大小倒序排序

 

-- 按照小时粒度统计，查询结果存入：( "dw_pvs_referer_h" )

drop table  dw_pvs_referer_h;

create table  dw_pvs_referer_h(referer_url string,referer_host string,month string,day  string,hour string,pv_referer_cnt bigint) partitioned by(datestr string);

 

insert into table  dw_pvs_referer_h partition(datestr='2016-03-18')

select  http_referer,ref_host,month,day,hour,count(1) as pv_referer_cnt

from ods_weblog_detail

group by  http_referer,ref_host,month,day,hour

having ref_host is not  null

order by hour asc,day  asc,month asc,pv_referer_cnt desc;

 

 

按天粒度统计各来访域名的访问次数并排序

drop table  dw_ref_host_visit_cnts_h;

create table dw_ref_host_visit_cnts_h(ref_host  string,month string,day string,hour string,ref_host_cnts bigint) partitioned  by(datestr string);

 

insert into table  dw_ref_host_visit_cnts_h partition(datestr='2016-03-18')

select  ref_host,month,day,hour,count(1) as ref_host_cnts

from ods_weblog_detail

group by  ref_host,month,day,hour

having ref_host is not  null

order by hour asc,day  asc,month asc,ref_host_cnts desc;

注：还可以按来源地域维度、访客终端维度等计算

 

1.4 统计pv总量最大的来源TOPN

需求描述：按照时间维度，比如，统计一天内产生最多pvs的来源topN

 

需要用到row_number函数

以下语句对每个小时内的来访host次数倒序排序标号，

selectref_host,ref_host_cnts,concat(month,hour,day),

row_number() over(partition by concat(month,hour,day) order by ref_host_cnts desc) as od

from dw_ref_host_visit_cnts_h

效果如下：

 

 

根据上述row_number的功能，可编写Hql取各小时的ref_host访问次数topn

drop table dw_pvs_refhost_topn_h;

create table  dw_pvs_refhost_topn_h(

hour string,

toporder string,

ref_host string,

ref_host_cnts string

) partitioned by(datestr  string);

 

insert into table  zs.dw_pvs_refhost_topn_h partition(datestr='2016-03-18')

select t.hour,t.od,t.ref_host,t.ref_host_cnts  from

 (select  ref_host,ref_host_cnts,concat(month,day,hour) as hour,

row_number() over  (partition by concat(month,day,hour) order by ref_host_cnts desc) as od

from  zs.dw_ref_host_visit_cnts_h) t where od<=3;

 

 

结果如下：

注：还可以按来源地域维度、访客终端维度等计算

 

2.    受访分析

统计每日最热门的页面top10

drop table dw_pvs_d;

create table dw_pvs_d(day  string,url string,pvs string);

 

insert into table  dw_pvs_d

select  '2013-09-18',a.request,a.request_counts from

(select request as  request,count(request) as request_counts from ods_weblog_detail where  datestr='2013-09-18' group by request having request is not null) a

order by a.request_counts  desc limit 10;

 

结果如下：

 

注：还可继续得出各维度交叉结果

 

 3.    访客分析

3.1 独立访客

需求描述：按照时间维度比如小时来统计独立访客及其产生的pvCnts

对于独立访客的识别，如果在原始日志中有用户标识，则根据用户标识即很好实现;

此处，由于原始日志中并没有用户标识，以访客IP来模拟，技术上是一样的，只是精确度相对较低

 

时间维度：时

drop table  dw_user_dstc_ip_h;

create table  dw_user_dstc_ip_h(

remote_addr string,

pvs      bigint,

hour     string);

 

insert into table  dw_user_dstc_ip_h

select  remote_addr,count(1) as pvs,concat(month,day,hour) as hour

from ods_weblog_detail

Where  datestr='2013-09-18'

group by  concat(month,day,hour),remote_addr;

 

在此结果表之上，可以进一步统计出，每小时独立访客总数，每小时请求次数topn访客等

如每小时独立访客总数：

select count(1) as  dstc_ip_cnts,hour from dw_user_dstc_ip_h group by hour;

 

 

练习：

统计每小时请求次数topn的独立访客

时间维度：月

select  remote_addr,count(1) as counts,month

from ods_weblog_detail

group by  month,remote_addr;

 

时间维度：日

select  remote_addr,count(1) as counts,concat(month,day) as day

from ods_weblog_detail

Where dd='18/Sep/2013'

group by  concat(month,day),remote_addr;

 

 

 

注：还可以按来源地域维度、访客终端维度等计算

 

 

 

 

 

 

 

 

 

 

3.2 每日新访客

需求描述：将每天的新访客统计出来

实现思路：创建一个去重访客累积表，然后将每日访客对比累积表

时间维度：日

--历日去重访客累积表

drop table  dw_user_dsct_history;

create table  dw_user_dsct_history(

day string,

ip string

)

partitioned by(datestr  string);

 

--每日新用户追加到累计表

drop table  dw_user_dsct_history;

create table  dw_user_dsct_history(

day string,

ip string

)

partitioned by(datestr  string);

 

--每日新用户追加到累计表

insert into table  dw_user_dsct_history partition(datestr='2013-09-19')

select tmp.day as  day,tmp.today_addr as new_ip from

(

select today.day as  day,today.remote_addr as today_addr,old.ip as old_addr

from

(select distinct  remote_addr as remote_addr,"2013-09-19" as day from  ods_weblog_detail where datestr="2013-09-19") today

left outer join

dw_user_dsct_history old

on  today.remote_addr=old.ip

) tmp

where tmp.old_addr is  null;

 

验证：

select count(distinct  remote_addr) from ods_weblog_detail;

-- 1005

 

select count(1) from  dw_user_dsct_history where prtflag_day='18/Sep/2013';

--845

 

select count(1) from  dw_user_dsct_history where prtflag_day='19/Sep/2013';

--160

 

 

时间维度：月

类似日粒度算法

 

注：还可以按来源地域维度、访客终端维度等计算

 

4.    Visit分析（点击流模型）

4.2 回头/单次访客统计

需求描述：查询今日所有回头访客及其访问次数

实现思路：上表中出现次数>1的访客，即回头访客；反之，则为单次访客

 

 

drop table  dw_user_returning;

create table  dw_user_returning(

day string,

remote_addr string,

acc_cnt string)

partitioned by (datestr  string);

 

insert overwrite table  dw_user_returning partition(datestr='2013-09-18')

 

select  tmp.day,tmp.remote_addr,tmp.acc_cnt

from

(select '2013-09-18' as  day,remote_addr,count(session) as acc_cnt from click_stream_visit group by  remote_addr) tmp

where tmp.acc_cnt>1;

 

4.3 人均访问频次

需求：统计出每天所有用户访问网站的平均次数（visit）

总visit数/去重总用户数

select sum(pagevisits)/count(distinct remote_addr) from  click_stream_visit partition(datestr='2013-09-18');

 

 

5. Visit分析另一种实现方式[dht1] 

5.1 mr程序识别出访客的每次访问

a.) 首先开发MAPREDUCE程序：UserStayTime

注：代码略长，见项目工程代码

                                                    

 

b.) 提交MAPREDUCE程序进行运算

[hadoop@hdp-node-01 ~]$  hadoop jar weblog.jar cn.itcast.bigdata.hive.mr.UserStayTime /weblog/input  /weblog/stayout4

--导入hive表("t_display_access_info")中

drop table  ods_access_info;

create table  ods_access_info(remote_addr string,firt_req_time string,last_req_time  string,stay_long string)

partitioned  by(prtflag_day string)

row format delimited

fields terminated by  '\t';

 

load data inpath  '/weblog/stayout4' into table ods_access_info  partition(prtflag_day='18/Sep/2013');

创建表时stay_long使用的string类型，但是在后续过程中发现还是用bigint更好，进行表修改

alter table  ods_access_info change column stay_long stay_long bigint;

 

5.2 将mr结果导入访客访问信息表"t_display_access_info"

由于有一些访问记录是单条记录，mr程序处理处的结果给的时长是0，所以考虑给单次请求的停留时间一个默认市场30秒

drop table  dw_access_info;

create table  dw_access_info(remote_addr string,firt_req_time string,last_req_time  string,stay_long string)

partitioned  by(prtflag_day string);

 

insert into table  dw_access_info partition(prtflag_day='19/Sep/2013')

select  remote_addr,firt_req_time,last_req_time,

case stay_long

when 0 then 30000

else stay_long

end as stay_long

from ods_access_info

where  prtflag_day='18/Sep/2013';

 

 

在访问信息表的基础之上，可以实现更多指标统计，如：

统计所有用户停留时间平均值，观察用户在站点停留时长的变化走势

select prtflag_day as dt,avg(stay_long) as avg_staylong

from dw_access_info group by prtflag_day;

 

 

 

5.3 回头/单次访客统计

注：从上一步骤得到的访问信息统计表“dw_access_info”中查询

 

--回头访客访问信息表 "dw_access_info_htip"

drop table  dw_access_info_htip;

create table  dw_access_info_htip(remote_addr string, firt_req_time string, last_req_time  string, stay_long string,acc_counts string)

partitioned  by(prtflag_day string);

 

insert into table  dw_access_info_htip partition(prtflag_day='18/Sep/2013')

select  b.remote_addr,b.firt_req_time,b.last_req_time,b.stay_long,a.acc_counts from

(select  remote_addr,count(remote_addr) as acc_counts from dw_access_info where  prtflag_day='18/Sep/2013' group by remote_addr having acc_counts>1) a

join

dw_access_info b

on a.remote_addr =  b.remote_addr;

 

 

--单次访客访问信息表 "dw_access_info_dcip"

drop table  dw_access_info_dcip;

create table  dw_access_info_dcip(remote_addr string, firt_req_time string, last_req_time  string, stay_long string,acc_counts string)

partitioned by(prtflag_day  string);

 

insert into table  dw_access_dcip partition(prtflag_day='18/Sep/2013')

select  b.remote_addr,b.firt_req_time,b.last_req_time,b.stay_long,a.acc_counts from

(select  remote_addr,count(remote_addr) as acc_counts from dw_access_info where  prtflag_day='18/Sep/2013' group by remote_addr having acc_counts<2) a

join

dw_access_info b

on a.remote_addr =  b.remote_addr;

 

 

 

 

 

在回头/单词访客信息表的基础之上，可以实现更多统计指标，如：

 

--当日回头客占比

drop table  dw_htpercent_d;

create table  dw_htpercent_d(day string,ht_percent float);

 

Insert into table  dw_htpercent_d

select  '18/Sep/2013',(tmp_ht.ht/tmp_all.amount)*100 from

(select count( distinct  a.remote_addr) as ht from dw_access_info_htip a where  prtflag_day='18/Sep/2013') tmp_ht

Join

(select count(distinct  b.remote_addr) as amount from dw_access_info b where  prtflag_day='18/Sep/2013') tmp_all;

 

 

 

 

 

 

5.4 人均访问频度

--总访问次数/去重总人数，从访客次数汇总表中查询

select  avg(user_times.counts) as user_access_freq

from

(select  remote_addr,counts from t_display_htip

union all

select remote_addr,counts  from t_display_access_dcip) user_times;

 

--或直接从访问信息表  t_display_access_info 中查询

select avg(a.acc_cts)  from

(select  remote_addr,count(*) as acc_cts from dw_access_info group by remote_addr) a;

 

 

 

 

 

 

 

 

 

 

 

 

6.关键路径转化率分析——漏斗模型

转化：在一条指定的业务流程中，各个步骤的完成人数及相对上一个步骤的百分比

6.1 需求分析

 

 

6.2 模型设计

定义好业务流程中的页面标识，下例中的步骤为：

Step1、 /item%

Step2、 /category

Step3、 /order

Step4、 /index

 

6.3 开发实现

分步骤开发：

 

1、查询每一个步骤的总访问人数

create table route_numbs  as

select 'step1' as  step,count(distinct remote_addr)  as  numbs from ods_click_pageviews where request like '/item%'

union

select 'step2' as  step,count(distinct remote_addr) as numbs from ods_click_pageviews where  request like '/category%'

union

select 'step3' as  step,count(distinct remote_addr) as numbs from ods_click_pageviews where  request like '/order%'

union

select 'step4' as  step,count(distinct remote_addr)  as  numbs from ods_click_pageviews where request like '/index%';

 

2、查询每一步骤相对于路径起点人数的比例

 

思路：利用join

select rn.step as  rnstep,rn.numbs as rnnumbs,rr.step as rrstep,rr.numbs as rrnumbs  from route_num rn

inner join

route_num rr

 

 

 

 

select  tmp.rnstep,tmp.rnnumbs/tmp.rrnumbs as ratio

from

(

select rn.step as  rnstep,rn.numbs as rnnumbs,rr.step as rrstep,rr.numbs as rrnumbs  from route_num rn

inner join

route_num rr) tmp

where tmp.rrstep='step1';

 

3、查询每一步骤相对于上一步骤的漏出率

select tmp.rrstep as  rrstep,tmp.rrnumbs/tmp.rnnumbs as ration

from

(

select rn.step as  rnstep,rn.numbs as rnnumbs,rr.step as rrstep,rr.numbs as rrnumbs  from route_num rn

inner join

route_num rr) tmp

where  cast(substr(tmp.rnstep,5,1) as int)=cast(substr(tmp.rrstep,5,1) as int)-1

 

 

4、汇总以上两种指标

select  abs.step,abs.numbs,abs.ratio as abs_ratio,rel.ratio as rel_ratio

from

(

select tmp.rnstep as  step,tmp.rnnumbs as numbs,tmp.rnnumbs/tmp.rrnumbs as ratio

from

(

select rn.step as  rnstep,rn.numbs as rnnumbs,rr.step as rrstep,rr.numbs as rrnumbs  from route_num rn

inner join

route_num rr) tmp

where tmp.rrstep='step1'

) abs

left outer join

(

select tmp.rrstep as  step,tmp.rrnumbs/tmp.rnnumbs as ratio

from

(

select rn.step as  rnstep,rn.numbs as rnnumbs,rr.step as rrstep,rr.numbs as rrnumbs  from route_num rn

inner join

route_num rr) tmp

where  cast(substr(tmp.rnstep,5,1) as int)=cast(substr(tmp.rrstep,5,1) as int)-1

) rel

on abs.step=rel.step

 

 

 

 

 

 

 

8 模块开发——结果导出

报表统计结果，由sqoop从hive表中导出，示例如下，详见工程代码

sqoop export \

--connect  jdbc:mysql://hdp-node-01:3306/webdb --username root --password root  \

--table  click_stream_visit  \

--export-dir  /user/hive/warehouse/dw_click.db/click_stream_visit/datestr=2013-09-18 \

--input-fields-terminated-by  '\001'

 

9 模块开发——工作流调度

注：将整个项目的数据处理过程，从数据采集到数据分析，再到结果数据的导出，一系列的任务分割成若干个oozie的工作流，并用coordinator进行协调

 

工作流定义示例

Ooize配置片段示例，详见项目工程

1、日志预处理mr程序工作流定义

<workflow-app  name="weblogpreprocess"  xmlns="uri:oozie:workflow:0.4">

<start  to="firstjob"/>

<action  name="firstjob">

<map-reduce>

<job-tracker>${jobTracker}</job-tracker>

<name-node>${nameNode}</name-node>

<prepare>

<delete  path="${nameNode}/${outpath}"/>

</prepare>

<configuration>

<property>

<name>mapreduce.job.map.class</name>

<value>cn.itcast.bigdata.hive.mr.WeblogPreProcess$WeblogPreProcessMapper</value>

</property>

 

<property>

<name>mapreduce.job.output.key.class</name>

<value>org.apache.hadoop.io.Text</value>

</property>

<property>

<name>mapreduce.job.output.value.class</name>

<value>org.apache.hadoop.io.NullWritable</value>

</property>

 

<property>

<name>mapreduce.input.fileinputformat.inputdir</name>

<value>${inpath}</value>

</property>

<property>

<name>mapreduce.output.fileoutputformat.outputdir</name>

<value>${outpath}</value>

</property>

<property>

<name>mapred.mapper.new-api</name>

<value>true</value>

</property>

<property>

<name>mapred.reducer.new-api</name>

<value>true</value>

</property>

 

</configuration>

</map-reduce>

<ok  to="end"/>

<error  to="kill"/>

 

2、数据加载etl工作流定义：

<workflow-app  xmlns="uri:oozie:workflow:0.5" name="hive2-wf">

<start  to="hive2-node"/>

 

<action  name="hive2-node">

<hive2  xmlns="uri:oozie:hive2-action:0.1">

<job-tracker>${jobTracker}</job-tracker>

<name-node>${nameNode}</name-node>

<configuration>

<property>

<name>mapred.job.queue.name</name>

<value>${queueName}</value>

</property>

</configuration>

<jdbc-url>jdbc:hive2://hdp-node-01:10000</jdbc-url>

<script>script.q</script>

<param>input=/weblog/outpre2</param>

</hive2>

<ok  to="end"/>

<error  to="fail"/>

</action>

 

<kill  name="fail">

<message>Hive2  (Beeline) action failed, error  message[${wf:errorMessage(wf:lastErrorNode())}]</message>

</kill>

<end  name="end"/>

</workflow-app>

 

3、数据加载工作流所用hive脚本：

create database if not  exists dw_weblog;

use dw_weblog;

drop table if exists  t_orgin_weblog;

create table  t_orgin_weblog(valid string,remote_addr string,

remote_user string,

time_local string,

request string,

status string,

body_bytes_sent string,

http_referer string,

http_user_agent string)

row format delimited

fields terminated by  '\001';

load data inpath  '/weblog/preout' overwrite into table t_orgin_weblog;

 

drop table if exists  t_ods_detail_tmp_referurl;

create table  t_ods_detail_tmp_referurl as

SELECT a.*,b.*

FROM t_orgin_weblog a

LATERAL VIEW  parse_url_tuple(regexp_replace(http_referer, "\"",  ""), 'HOST', 'PATH','QUERY', 'QUERY:id') b as host, path, query,  query_id;

 

drop table if exists  t_ods_detail;

create table t_ods_detail  as

select  b.*,substring(time_local,0,11) as daystr,

substring(time_local,13)  as tmstr,

substring(time_local,4,3)  as month,

substring(time_local,0,2)  as day,

substring(time_local,13,2)  as hour

from  t_ods_detail_tmp_referurl b;

 

drop table  t_ods_detail_prt;

create table  t_ods_detail_prt(

valid                  string,

remote_addr            string,

remote_user            string,

time_local             string,

request                string,

status                 string,

body_bytes_sent        string,

http_referer           string,

http_user_agent        string,

host                   string,

path                   string,

query                  string,

query_id               string,

daystr                 string,

tmstr                  string,

month                  string,

day                    string,

hour                   string)

partitioned by (mm  string,dd string);

 

 

insert into table  t_ods_detail_prt partition(mm='Sep',dd='18')

select * from  t_ods_detail where daystr='18/Sep/2013';

insert into table  t_ods_detail_prt partition(mm='Sep',dd='19')

select * from  t_ods_detail where daystr='19/Sep/2013';

 

 

更多工作流及hql脚本定义详见项目工程

---

下节是单元测试，和可视化展示。

听说有气质的人都会关注这个公众号！