日志处理流程：Flume+MapReduce+Hive+Sqoop+MySQL

（1）使用Flume-ng的spooldir类型的source来监控Linux文件系统上的一个目录，并使用hdfs类型的sink将日志数据传输到HDFS。以下是Flume-ng agent的配置文件a4.conf的内容：

# 定义agent名，source、channel、sink的名称
a4.sources = r1
a4.channels = c1
a4.sinks = k1

# 具体定义source
a4.sources.r1.type = spooldir
a4.sources.r1.spoolDir = /root/Documents/logs

# 具体定义channel
a4.channels.c1.type = memory
a4.channels.c1.capacity = 10000
a4.channels.c1.transactiOnCapacity= 100

# 定义拦截器，为消息添加时间戳
a4.sources.r1.interceptors = i1
a4.sources.r1.interceptors.i1.type = org.apache.flume.interceptor.TimestampInterceptor$Builder

# 具体定义sink
a4.sinks.k1.type = hdfs
a4.sinks.k1.hdfs.path = hdfs://hadoop:9000/flume/%Y%m%d
a4.sinks.k1.hdfs.filePrefix = events-
a4.sinks.k1.hdfs.fileType = DataStream
# 不按照条数生成文件
a4.sinks.k1.hdfs.rollCount = 0
# HDFS上的文件达到128M时生成一个文件
a4.sinks.k1.hdfs.rollSize = 134217728
# HDFS上的文件达到60秒生成一个文件
a4.sinks.k1.hdfs.rollInterval = 60

# 组装source、channel、sink
a4.sources.r1.channels = c1
a4.sinks.k1.channel = c1

在Flume的根目录下，使用以下命令启动Flume监听： bin/flume-ng agent -n a4 -c conf -f conf/a4.conf -Dflume.root.logger=INFO,console。一旦目标目录下新增内容，这些内容将被自动收集到HDFS的指定目录。

（2）使用MapReduce、Hive、Sqoop和MySQL进行数据处理。首先，创建一个Hive外部分区表bbslog，指向清洗后的数据目录bbslog_cleaned。接着，使用MapReduce程序清洗位于/flume目录下的原始日志数据，并将结果保存到/bbslog_cleaned目录。然后，告知bbslog表新的分区，分析清洗后的日志数据，计算PV、UV、新注册用户数和VIP用户，并将结果分别保存到四个Hive表中。最后，使用Sqoop将这四个Hive表中的数据导出到MySQL的bbslog_out库中的相应表中。以下是执行脚本daily.sh的内容：

CURRENT=`date +%Y%m%d`

# 使用MapReduce程序清洗日志数据
/usr/hadoop/hadoop-2.2.0/bin/hadoop jar /root/Documents/cleaner.jar /flume/$CURRENT /bbslog_cleaned/$CURRENT

# 为Hive外部分区表bbslog添加分区
/usr/hive/apache-hive-0.13.0-bin/bin/hive -e "alter table bbslog add partition(logdate=$CURRENT) location '/bbslog_cleaned/$CURRENT'"

# 计算当天的PV
/usr/hive/apache-hive-0.13.0-bin/bin/hive -e "create table pv_$CURRENT row format delimited fields terminated by '\t' as select $CURRENT, count(*) from bbslog where logdate=$CURRENT"

# 计算当天的UV
/usr/hive/apache-hive-0.13.0-bin/bin/hive -e "create table uv_$CURRENT row format delimited fields terminated by '\t' as select $CURRENT, count(distinct ip) from bbslog where logdate=$CURRENT"

# 计算当天的新注册用户数
/usr/hive/apache-hive-0.13.0-bin/bin/hive -e "create table newregister_$CURRENT row format delimited fields terminated by '\t' as select $CURRENT, count(*) from bbslog where logdate=$CURRENT and instr(url, 'member.php?mod=register')>0"

# 计算当天的VIP用户
/usr/hive/apache-hive-0.13.0-bin/bin/hive -e "create table vip_$CURRENT row format delimited fields terminated by '\t' as select $CURRENT, ip, count(*) as hits from bbslog where logdate=$CURRENT group by ip having hits > 20 order by hits desc limit 10"

# 将pv_$CURRENT表中的数据导出到MySQL的pv表
/usr/sqoop/sqoop-1.4.4.bin__hadoop-2.0.4-alpha/bin/sqoop export --connect jdbc:mysql://192.168.1.10:3306/bbslog_out --username root --password 123 --export-dir "/user/hive/warehouse/pv_$CURRENT" --table pv --fields-terminated-by '\t'

# 将uv_$CURRENT表中的数据导出到MySQL的uv表
/usr/sqoop/sqoop-1.4.4.bin__hadoop-2.0.4-alpha/bin/sqoop export --connect jdbc:mysql://192.168.1.10:3306/bbslog_out --username root --password 123 --export-dir "/user/hive/warehouse/uv_$CURRENT" --table uv --fields-terminated-by '\t'

# 将newregister_$CURRENT表中的数据导出到MySQL的新注册用户表
/usr/sqoop/sqoop-1.4.4.bin__hadoop-2.0.4-alpha/bin/sqoop export --connect jdbc:mysql://192.168.1.10:3306/bbslog_out --username root --password 123 --export-dir "/user/hive/warehouse/newregister_$CURRENT" --table newRegister --fields-terminated-by '\t'

# 将vip_$CURRENT表中的数据导出到MySQL的vip表
/usr/sqoop/sqoop-1.4.4.bin__hadoop-2.0.4-alpha/bin/sqoop export --connect jdbc:mysql://192.168.1.10:3306/bbslog_out --username root --password 123 -m 1 --export-dir "/user/hive/warehouse/vip_$CURRENT" --table vip --fields-terminated-by '\t'