Kettle与Hadoop（五）执行MapReduce
Kettle与Hadoop（五）执行MapReduce
一、示例1——格式化原始web日志
本示例说明如何使用Pentaho MapReduce把原始web日志解析成格式化的记录。
1. 准备文件与目录
# 创建原始文件所在目录
hdfs dfs -mkdir /user/root/raw
# 修改读写权限
hdfs dfs -chmod -R 777 /用Hadoop copy files作业项将weblogs_rebuild.txt文件放到HDFS的/user/root/raw目录下，具体操作参见https://wxy0327.blog.csdn.net/article/details/106471124#%E4%B8%80%E3%80%81%E5%90%91Hadoop%E9%9B%86%E7%BE%A4%E5%AF%BC%E5%85%A5%E6%95%B0%E6%8D%AE%EF%BC%88Hadoop%20copy%20files%EF%BC%89。
2. 建立一个用于Mapper的转换
（1）新建一个转换，如图1所示。
图1
（2）编辑'MapReduce Input'步骤，如图2所示。
图2
（3）编辑'Regex Evaluation'步骤，如图3所示。
图3
说明：
- “Regular expression”里面填写如下内容：
^([^\s]{7,15})\s # client_ip
-\s # unused IDENT field
-\s # unused USER field
\[((\d{2})/(\w{3})/(\d{4}) # request date dd/MMM/yyyy
:(\d{2}):(\d{2}):(\d{2})\s([-+ ]\d{4}))\]
# request time :HH:mm:ss -0800
\s"(GET|POST)\s # HTTP verb
([^\s]*) # HTTP URI
\sHTTP/1\.[01]"\s # HTTP version
(\d{3})\s # HTTP status code
(\d+)\s # bytes returned
"([^"]+)"\s # referrer field
" # User agent parsing, always quoted.
"? # Sometimes if the user spoofs the user_agent, they incorrectly quote it.
( # The UA string
[^"]*? # Uninteresting bits
(?:
(?:
rv: # Beginning of the gecko engine version token
(?=[^;)]{3,15}[;)]) # ensure version string size
( # Whole gecko version
(\d{1,2}) # version_component_major
\.(\d{1,2}[^.;)]{0,8}) # version_component_minor
(?:\.(\d{1,2}[^.;)]{0,8}))? # version_component_a
(?:\.(\d{1,2}[^.;)]{0,8}))? # version_component_b
)
[^"]* # More uninteresting bits
)
|
[^"]* # More uninteresting bits
)
) # End of UA string
"?
"- “Capture Group Fields”如下所示，所有字段都是String类型。
client_ip
full_request_date
day
month
year
hour
minute
second
timezone
http_verb
uri
http_status_code
bytes_returned
referrer
user_agent
firefox_gecko_version
firefox_gecko_version_major
firefox_gecko_version_minor
firefox_gecko_version_a
firefox_gecko_version_b（4）编辑'Filter Rows'步骤，如图4所示。
图4
（5）编辑'Value Mapper'步骤，如图5所示。
图5
（6）编辑'User Defined Java Expression'步骤，如图6所示。
图6
说明：“Java Expression”列填写如下内容：
client_ip + '\t' + full_request_date + '\t' + day + '\t' + month + '\t' + month_num + '\t' + year + '\t' + hour + '\t' + minute + '\t' + second + '\t' + timezone + '\t' + http_verb + '\t' + uri + '\t' + http_status_code + '\t' + bytes_returned + '\t' + referrer + '\t' + user_agent（7）编辑'MapReduce Output'步骤，如图7所示。
图7
将转换保存为weblog_parse_mapper.ktr。
3. 建立一个调用MapReduce步骤的作业，使用mapper转换，仅运行map作业。
（1）新建一个作业，如图8所示。
图8
（2）编辑'Pentaho MapReduce'作业项，如图9到图11所示。
图9
图10
图11
说明：
- 只需要编辑“Mapper”、“Job Setup”和“Cluster”三个标签。
- CDH631是已经建立好的Hadoop Clusters，建立过程参见“https://wxy0327.blog.csdn.net/article/details/106406702#%E4%BA%8C%E3%80%81%E8%BF%9E%E6%8E%A5Hadoop%E9%9B%86%E7%BE%A4”。
将作业保存为weblogs_parse_mr.kjb。
4. 执行作业并验证输出
（1）执行作业，日志如图12所示。
图12
从图12可以看到，作业已经成功执行。
（2）检查HDFS的输出文件，结果如图13所示。
图13
从图13可以看到，/user/root/parse目录下生成了名为part-00000和part-00001的两个输出文件，内容已经被格式化。
二、示例2——生成聚合数据集
本示例说明如何使用Pentaho MapReduce把细节数据转换和汇总成一个聚合数据集。当给一个关系型数据仓库或数据集市准备待抽取的数据时，这是一个常见使用场景。我们使用格式化的web日志数据作为细节数据，并且建立一个聚合文件，包含按IP和年月分组的PV数。
1. 准备文件与目录
# 创建格式化文件所在目录
hdfs dfs -mkdir /user/root/parse1/
# 上传格式化文件
hdfs dfs -put -f weblogs_parse.txt /user/root/parse1/
# 修改读写权限
hdfs dfs -chmod -R 777 /2. 建立一个用于Mapper的转换
（1）新建一个转换，如图14所示。
图14
（2）编辑'MapReduce Input'步骤，如图15所示。
图15
（3）编辑'Split Fields'步骤，如图16所示。
图16
说明：“New field”如下所示，所有字段都是String类型。
client_ip
full_request_date
day
month
month_num
year
hour
minute
second
timezone
http_verb
uri
http_status_code
bytes_returned
referrer
user_agent（4）编辑'User Defined Java Expression'步骤，如图17所示。
图17
说明：“Java Expression”列填写如下内容：
client_ip + '\t' + year + '\t' + month_num（5）编辑'MapReduce Output'步骤，如图18所示。
图18
将转换保存为aggregate_mapper.ktr。
3. 建立一个用于Reducer的转换
（1）新建一个转换，如图19所示。
图19
（2）编辑'MapReduce Input'步骤，如图20所示。
图20
（3）编辑'Group by'步骤，如图21所示。
图21
（4）编辑'MapReduce Output'步骤，如图22所示。
图22
将转换保存为aggregate_reducer.ktr。
4. 建立一个调用MapReduce步骤的作业，调用mapper和reducer转换。
（1）新建一个作业，如图23所示。
图23
（2）编辑'Pentaho MapReduce'作业项，如图24到图27所示。
图24
图25
图26
图27
说明：
- 需要编辑“Mapper”、“Reducer”、“Job Setup”和“Cluster”四个标签。
- CDH631是已经建立好的Hadoop Clusters。
将作业保存为aggregate_mr.kjb。
5. 执行作业并验证输出
（1）执行作业，日志如图28所示。
图28
从图28可以看到，作业已经成功执行。
（2）检查HDFS的输出文件，结果如图29所示。
图29
从图29可以看到，/user/root/aggregate_mr目录下生成了名为part-00000输出文件，文件中包含按IP和年月分组的PV数。