Flume案例——日志分析采集系统

Posted by Yezhiwei on April 25, 2018

概述

大数据平台每天处理业务系统产生的大量日志数据,一般而言,这些系统需要具有以下特征:

  1. 构建业务系统和日志分析系统的桥梁,并将它们之间的关联解耦;
  2. 支持近实时的在线分析系统和类似于 Hadoop 之类的离线分析系统;
  3. 具有高可扩展性。即:当数据量增加时,可以通过增加节点进行水平扩展。

日志分析采集系统模块

日志分析采集系统模块

  • 数据采集模块:负责从各节点上实时采集数据,选用 Flume-NG 来实现。 Flume-NG 提供了丰富的 Source、Channel、Sink 实现,各种数据源的引入只要变更配置就可实现。
  • 数据接入模块:由于采集数据的速度和数据处理的速度不一定同步(类似于生产者消费者模式),因此添加一个消息中间件来作为缓冲,这里选用 Kafka 来实现。Kafka 适用于对数据管道的吞吐量、可用性要求都很高的解决方案,需要编程实现数据的生产和消费。
  • 流式计算模块:对采集到的数据进行实时分析,选用 Storm 来实现。
  • 数据输出模块:对分析后的结果持久化,可以使用 HDFS、MySQL 等。

采用 Flume 作为数据的生产者,将生产的消息数据(日志数据、业务请求数据等)通过 Kafka Sink 发布到 Kafka 中。然后再由流式计算程序 Storm 做实时分析,这时就需要将在 Storm 的 Spout 中读取 Kafka 中的消息,然后交由具体的 Spot 组件去分析处理。同时 Storm 已经自带了一个集成 Kafka 的外部插件程序 storm-kafka。最后将计算结果持久化到 Hadoop 或 MySQL 中。

示例配置

Kafka 集群配置

  • 使用3台机器搭建 Kafka 集群,在每台机器上添加如下配置:
# vim /etc/hosts
192.168.111.238 master
192.168.111.239 slave1
192.168.111.240 slave2
  • 在安装 Kafka 集群时,没有使用 Kafka 自带的 Zookeeper,而是独立安装了一个 Zookeeper 集群,也是使用这3台机器,并且保证 Zookeeper 集群正常运行。
  • 在 master 上准备 Kafka 安装文件,执行如下命令:
cd /usr/local/
tar xvzf kafka_2.11-1.0.0.tgz
ln -s /usr/local/kafka_2.11-1.0.0 /usr/local/kafka
  • 修改配置文件 /usr/local/kafka/config/server.properties
broker.id=0
zookeeper.connect=master:2181,slave1:2181,slave2:2181/kafka

说明 :

默认 Kafka 会使用 ZooKeeper 默认的 / 路径,导致有关 Kafka 的 ZooKeeper 配置就会散落在根路径下面,如果有其他的应用也在使用此 ZooKeeper 集群,查看 ZooKeeper 中的数据会不直观,所以指定一个 chroot 路径,直接在 zookeeper.connect 配置项中指定

  • 需要手动在 ZooKeeper 中创建路径 /kafka,使用如下命令连接到任意一台 ZooKeeper 服务器:
cd /usr/local/zookeeper
bin/zkCli.sh
create /kafka ''
  • 将配置好的安装文件同步到其他的 slave1、slave2 节点上:
scp -r /usr/local/kafka_2.11-1.0.0/ slave1:/usr/local/
scp -r /usr/local/kafka_2.11-1.0.0/ slave2:/usr/local/
  • 分别修改配置文件/usr/local/kafka/config/server.properties内容如下
# 因为Kafka集群需要保证各个Broker的id在整个集群中必须唯一,需要调整这个配置项的值
broker.id=1  # 在slave1修改
 
broker.id=2  # 在slave2修改
  • 在集群中的 master、slave1、slave2 这三个节点上分别启动 Kafka,分别执行如下命令:
bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &
  • 通过查看日志,或者检查进程状态,保证 Kafka 集群启动成功。

配置 Flume agent,将 Flume 与 Kafka 整合, 配置文件为 flume-conf.properties

agent.sources = logsrc
agent.channels = memchannel
agent.sinks = kafkasink

# configure the source
agent.sources.logsrc.type = exec
agent.sources.logsrc.command = tail -F /data/logs/component_role.log
agent.sources.logsrc.shell = /bin/sh -c
agent.sources.logsrc.batchSize = 50
agent.sources.logsrc.channels = memchannel

# configure the sink
agent.sinks.kafkasink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafkasink.brokerList=master:9092, slave1:9092,slave2:9092
agent.sinks.kafkasink.topic=mytopic
agent.sinks.kafkasink.requiredAcks = 1
agent.sinks.kafkasink.batchSize = 20
agent.sinks.kafkasink.channel = memchannel

# configure the channel
agent.channels.memchannel.type = memory
agent.channels.memchannel.capacity = 1000

启动该 Flume 节点

cd /usr/local/apache-flume-1.7.0-bin
./bin/flume-ng agent -n agent -c ./conf -f ./conf/flume-conf.properties -Dflume.monitoring.type=http -Dflume.monitoring.port=10100 -Dflume.root.logger=DEBUG,console

动态追加日志数据,模拟日志输出,进行测试

动态追加日志数据,执行命令向 /data/logs/component_role.log 添加数据:

echo  "测试代码" >>  /data/logs/component_role.log
echo  "检测Flume+Kafka数据管道通畅" >>  /data/logs/component_role.log

验证 Kafka 数据接收结果

root@ubuntu238:/usr/local/kafka_2.11-1.0.0# ./bin/kafka-console-consumer.sh --zookeeper master:2181,slave1:2181,slave2:2181/kafka  --topic mytopic --from-beginning
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
测试代码
检测Flume+Kafka数据管道通畅

未完,后续增加流式计算及数据输出。