26个数据分析案例——第四站：基于Flume与Kafka的web服务器日志数据采集
实验环境
- Python: Python 3.x；
- Hadoop2.7.2环境；
- Kafka_2.11;
- Flume-1.9.0.
资料包
链接：https://pan.baidu.com/s/1oZcqAx0EIRF7Aj1xxm3WNw
提取码：kohe
实验步骤
第一步：安装并启动httpd服务器
[root@master ~]# yum -y install httpd
[root@master ~]# cd /var/www/html/
[root@master html]# vi index.html #输入如下内容
Hello Flume
[root@master html]# service httpd start
通过浏览器访问结果为。代表启动成功。
查看是否有日志生成。
[root@master html]# cd /var/log/httpd/
[root@master httpd]#cat access_log
结果为。
第二步：配置Flume。
进入/usr/local/flume/conf目录，在该目录下创建名为access_log-HDFS.properties的配置文件。设置监控/var/log/httpd/目录下的access_log日志文件，并将内容通过9092端口发送至kafka。
[root@master httpd]# cd /usr/local/flume/conf/
[root@master conf]# vi access_log-HDFS.properties
#配置文件内容如下所示。
a1.sources = s1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.s1.type = exec
a1.sources.s1.command = tail -f /var/log/httpd/access_log
a1.sources.s1.channels=c1
a1.sources.s1.fileHeader = false
a1.sources.s1.interceptors = i1
a1.sources.s1.interceptors.i1.type = timestamp
#Kafka sink配置
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = cmcc
a1.sinks.k1.brokerList = master:9092
a1.sinks.k1.requiredAcks = 1
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.s1.channels = c1
a1.sinks.k1.channel = c1
第三步：配置kafka
首先应确保zookeeper和kafka进程处于正常状态，若为关闭状态可使用以下命令启动。
[root@master ~]# /usr/local/zookeeper/bin/zkServer.sh start
[root@master ~]# cd /usr/local/kafka/bin/
[root@master bin]# ./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
第四步：编写代码使用Python消费kafka数据
这里使用PyKafkafka模块实现，创建名为kafkacustomer.py文件，用于从9092端口消费kafka数据，消费数据时需要制定topic，代码如下。
[root@master ~]# vim kafkacustomer.py
from pykafka import KafkaClient
client = KafkaClient(hosts="192.168.0.10:9092")
topic = client.topics['cmcc'] //指定从cmcc这个topic中消费数据
consumer = topic.get_simple_consumer(
consumer_group="tpic",
reset_offset_on_start=True
)
for message in consumer: //遍历接收到的内容
if message is not None: //如果信息不为空
print(message.offset, message.value) //数据结果
第五步：启动项目
启动Flume根目录使用access_log-HDFS.properties配置文件启动数据采集，数据采集启动后会自动创建kafka的topic名为cmcc。
[root@master ~]# cd /usr/local/flume/
[root@master flume]# bin/flume-ng agent --name a1 --conf conf --conf-file conf/access_log-HDFS.properties -Dflume.root.logger=INFO,console
使用Python3运行kafkacustomer.py文件，每当页面刷新时，会实时产生日志文件，并且由flume采集到发送到kafka。
[root@master ~]python customerkafka.py
后续案例持续更新
01 基于Python的HBase冠字号查询系统
02基于Hive的民航客户价值分析
03 基于python的药店销售数据分析
04基于Flume与Kafka的web服务器日志数据采集
05 慕课网数据采集并处理
06 Linux操作系统实时日志采集并处理
07 医疗行业案例-中医病症辩证关联规则分析
08教育行业案例-高校学生生活数据分析
10 娱乐行业案例-广告收益回归预测模型
11网络行业案例-网站访问行为分析
12零售行业案例-店铺热门商品实时统计
13 营业额数据可视化
14金融行业案例-基于上市公司股票信息及其衍生变量的金融数据分析
15银行信用卡风险数据可视化
16 滴滴出行城市运营分析
17 幸福指数可视化
18 员工主动离职预警模型
19 歌手推荐模型
20 2020新冠肺炎疫情数据分析
21 淘宝购物狂欢节数据分析
22 共享单车数据分析
23人脸检测系统
24服装分拣系统
25口罩佩戴识别系统
26 imdb电影数据分析