Flume简介与使用——Kafka Sink消费数据之Kafka安装

Posted WOTGL

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Flume简介与使用——Kafka Sink消费数据之Kafka安装相关的知识,希望对你有一定的参考价值。

  前面已经介绍了如何利用Thrift Source生产数据,今天介绍如何用Kafka Sink消费数据。

  其实之前已经在Flume配置文件里设置了用Kafka Sink消费数据

agent1.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink
agent1.sinks.kafkaSink.topic = TRAFFIC_LOG
agent1.sinks.kafkaSink.brokerList = 10.208.129.3:9092,10.208.129.4:9092,10.208.129.5:9092
agent1.sinks.kafkaSink.metadata.broker.list = 10.208.129.3:9092,10.208.129.4:9092,10.208.129.5:9092
agent1.sinks.kafkaSink.producer.type=sync
agent1.sinks.kafkaSink.serializer.class=kafka.serializer.DefaultEncoder
agent1.sinks.kafkaSink.channel = memoryChannel

  那么当Flume的channel收到数据的时候,会根据配置文件主动把数据event发送到Kafka的broker上,所以只要安装好Kafka就可以消费收据了。

Step 1: Download the code

下载安装包并解压

 > tar -xzf kafka_2.11-0.10.0.0.tgz

 > cd kafka_2.11-0.10.0.0

Step 2: Start the server

Kafka是基于Zookeeperl来实现分布式协同的,因此先启动Zookeeper:

 > %Zookeeper_Home%/bin/zkServer.sh start

在配置文件server.properties把下面一句前面的注释去掉,然后启动Kafka服务器

 > #listeners=PLAINTEXT://:9092

 > bin/kafka-server-start.sh config/server.properties

接下来启动其他两个broker:

 > cp config/server.properties config/server-1.properties

 > cp config/server.properties config/server-2.properties

 修改配置文件,broker.id不能重复

config/server-1.properties:
    broker.id=1
config/server-2.properties: broker.id=2

Step 3: Create a topic

创建一个TRAFFIC_LOG主题的broker,复制因子为3(因为有3个Kafka服务器集群),分区个数为1

 > bin/kafka-topics.sh --create --zookeeper 10.208.129.4:2181 --replication-factor 3 --partitions 1 --topic TRAFFIC_LOG

Step 5: Start a consumer

 > bin/kafka-console-consumer.sh --zookeeper 10.208.129.4:2181 --topic TRAFFIC_LOG --from-beginning

如果在终端看到之前接入的Thrift Source输出,那么整个Flume+Kafka算是跑通了



这里已经引入了Flume和Kafka,下一篇将介绍Kafka以及Flume和Kafka的区别

以上是关于Flume简介与使用——Kafka Sink消费数据之Kafka安装的主要内容,如果未能解决你的问题,请参考以下文章

flume与kafka集成配置

[Flume][Kafka]Flume 与 Kakfa结合例子(Kakfa 作为flume 的sink 输出到 Kafka topic)

flume从Kafka消费数据到HDFS

尚硅谷电商数仓 Flume写入hdfs报错

尚硅谷电商数仓 Flume写入hdfs报错

尚硅谷电商数仓 Flume写入hdfs报错