Flume+Kafka+Zookeeper搭建大数据日志采集框架

1.  JDK的安装

    参考jdk的安装,此处略。

2.  安装Zookeeper

    参考我的Zookeeper安装教程中的“完全分布式”部分。

3.  安装Kafka

    参考我的Kafka安装教程中的“完全分布式搭建”部分。

4.  安装Flume

    参考我的Flume安装教程。

5.  配置Flume

5.1.  配置kafka-s.cfg

    $ cd /software/flume/conf/                   #切换到kafka的配置文件目录

    $ cp flume-conf.properties.template kafka-s.cfg   #flume的配置文件模板复制为kafka-s.cfg

 

    kafka-s.cfg的详细内容如下:

    #配置flume agentsourcechannelsink

    a1.sources = r1

    a1.channels = c1

    a1.sinks=k1

    #配置source

    a1.sources.r1.type = exec

    a1.sources.r1.command=tail -F/tmp/logs/kafka.log

 

    # 配置channel

    a1.channels.c1.type = memory

    a1.channels.c1.capacity = 1000

    a1.channels.c1.transactionCapacity = 100

    #配置sink

    a1.sinks.k1.channel = c1

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

    #设置KafkaTopic

    a1.sinks.k1.kafka.topic = mytest

    #设置Kafkabroker地址和端口号

    a1.sinks.k1.kafka.bootstrap.servers = s201:9092,s202:9092,s203:9092

    #配置批量提交的数量

    a1.sinks.k1.kafka.flumeBatchSize = 20

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

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

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

 

    # 绑定sourcesink channel

    a1.sources.r1.channels = c1

    a1.sinks.k1.channel=c1

 

    关于上面配置文件中需要注意的3点:

    aa1.sources.r1.command=tail-F /tmp/logs/kafka.log  

    ba1.sinks.k1.kafka.bootstrap.servers= s201:9092,s202:9092,s203:9092

    ca1.sinks.k1.topic= mytest

 

    由配置文件我们可以了解到:

    1)、我们需要在/tmp/logs下建一个kafka.log的文件,且向文件中输出内容(下面会说到);

    2)、flume连接到kafka的地址是 s201:9092,s202:9092,s203:9092,注意不要配置错误了;

    3)、flume会将采集后的内容输出到Kafkatopic mytest上,所以我们启动zkkafka后需要打开一个终端消费topic mytest的内容,这样就可以看到flumekafka之间玩起来了。

 

5.2.  创建/tmp/logs/kafka.log文件目录及文件

/tmp/logs下建立空文件kafka.log;如果/tmp目录下没有logs目录,则需要先创建logs目录。

5.3.  创建生成日志数据shell脚本

hadoop用户目录下创建kafkaoutput.sh脚本,并赋予执行权限,用来向/tmp/logs/kafka.log输出内容。

kafkaoutput.sh脚本的具体内容如下:

for((i=0;i<=1000;i++));

do echo”kafka_test-“+$i>>/tmp/logs/kafka.log;

done

5.4.  启动Zookeeper

    到zk安装的服务器上启动zk服务,命令如下:

    $ zkServer.sh start

5.5.  启动Kafka

    到部署kafka每台服务器上面启动kafka集群

    $ kafka-server-start.sh/software/kafka/config/server.properties &

5.6.  创建Kafka Topic

    $ kafka-topics.sh –create –zookeeper s201:2181–replication-factor 3 –partitions 3 –topic mytest

5.7.  启动消费Topic

    $ kafka-console-consumer.sh–bootstrap-server s201:9092,s202:9092,s203:9092 –zookeepers201:2181,s202:2181,s203:2181 –topic mytest –from-beginning

5.8.  启动Flume

    $ flume-ng agent –conf/software/flume/conf/ –conf-file kafka-s.cfg –name a1-Dflume.root.logger=INFO,console

5.9.  执行kafkaoutput.sh脚本生成日志数据

    $ ./kafkaoutput.sh

    查看日志文件内容如下:

Flume+Kafka+Zookeeper搭建大数据日志采集框架

kafka中查看的消费信息:

Flume+Kafka+Zookeeper搭建大数据日志采集框架

原创文章,作者:carmelaweatherly,如若转载,请注明出处:https://blog.ytso.com/194090.html

(0)
上一篇 2021年11月15日
下一篇 2021年11月15日

相关推荐

发表回复

登录后才能评论