日志项目实战(配置Kafka/flume)

一:　　配置Kafka

#:启动zookeeper

登录master、slave1、slave2上分别执行启动zookeeper命令:zkServer.sh start

#:启动Kafka

(首先要先进入到Kafka的安装目录:cd /export/servers/kafka_2.11-0.10.1.0/)

登录master、slave1、slave2上分别执行启动Kafka命令:kafka-server-start.sh -daemon ./config/server.properties

#:创建topic　　(存放网络日志的文件)

kafka-topics.sh --zookeeper master:2181,slave1:2181,slave2:2181 --create --topic aboutyunlog --replication-factor 1 --partitions 3

#:创建消费者

kafka-console-consumer.sh --zookeeper master:2181,slave1:2181,slave2:2181 --topic aboutyunlog --from-beginning

二:　　配置Agent

#:进入flume目录中新增配置文件

/export/servers/apache-flume-1.7.0-bin/conf
新增一个agent配置文件,命名:log_collection.conf,内容如下:

# agent的名称为a1

a1.sources = source1

a1.channels = channel1

a1.sinks = sink1

 

# set source

#a1.sources.source1.type = spooldir

a1.sources.source1.type = TAILDIR

a1.sources.source1.filegroups = f1

a1.sources.source1.filegroups.f1 = /opt/lampp/logs/access_log

a1sources.source1.fileHeader = flase

 

# set sink

a1.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSink

#a1.sinks.sink1.kafka.bootstrap.servers = master:9092,slave1:9092,slave2:9092

a1.sinks.sink1.brokerList= master:9092,slave1:9092,slave2:9092

a1.sinks.sink1.topic= aboutyunlog

a1.sinks.sink1.kafka.flumeBatchSize = 20

a1.sinks.sink1.kafka.producer.acks = 1

a1.sinks.sink1.kafka.producer.linger.ms = 1

a1.sinks.sink1.kafka.producer.compression.type = snappy

 

# set channel

a1.channels.channel1.type = file

a1.channels.channel1.checkpointDir = /home/aboutyun/data/flume_data/checkpoint

a1.channels.channel1.dataDirs= /home/aboutyun/data/flume_data/data

 

# bind

a1.sources.source1.channels = channel1

a1.sinks.sink1.channel = channel1

#:创建所依赖的目录

mkdir -p /home/aboutyun/data/flume_data/checkpoint
mkdir -p /home/aboutyun/data/flume_data/data

三:　　启动flume

#:进入到flume目录

cd /export/servers/apache-flume-1.7.0-bin/bin/
nohup ./flume-ng agent --conf-file /export/servers/apache-flume-1.7.0-bin/conf/log_collection.conf --name a1 -Dflume.root.logger=INFO,console &