Flume + Kafka学习——整合Demo

发表于 | 分类于
字数统计: 1,484字 | 阅读时长 ≈ 6 分钟

Flume + Kafka学习——整合Demo

环境

  1. Flume 1.6.0-cdh5.7.0
  2. Kafka 0.9.0.0

FLume官方文档
Flume1.6.0 User Guide
Kafka官网文档
Kafka Documentation

Flume

Apache Flume is a distributed, reliable, and available system for efficiently collecting, aggregating and moving large amounts of log data from many different sources to a centralized data store.

Flume是分布式的日志收集系统，它将各个服务器中的数据收集起来并送到目的地

Flume数据流模型

Flume就是将数据从数据源(source)收集过来，Flume会先缓存数据(channel),再将收集到的数据送到指定的目的地(sink)，最后Flume在删除自己缓存的数据

这样就是一个Event ，被定义为具有字节数组和可选字符串属性的数据流单元。包括 event headers、event body、event信息
数据流模型

Flume Agent 是一个（JVM）进程，用于承载Event从外部源流向下一个目标

Agent 由三个核心组成

  1. Source source组件是专门用来收集数据的,类似生产者。消耗由外部源（如Web服务器）传递给它的Event。外部源以Flume源能识别的格式向Flume发送Event
  2. Channel source组件把数据收集来以后临时存放在channel中，类似仓库
  3. Sink sink组件是用于把数据发送到目的地，类似消费者

Kafka

Kafka® is a distributed, partitioned, replicated commit log service. It provides the functionality of a messaging system, but with a unique design.

Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景

  1. producer：消息生产者，发布消息到 kafka 集群的终端或服务。
  2. broker：kafka 集群中包含的服务器。
  3. topic：每条发布到 kafka 集群的消息属于的类别，即 kafka 是面向 topic 的。
  4. partition：partition 是物理上的概念，每个 topic 包含一个或多个 partition。kafka 分配的单位是 partition。
  5. consumer：从 kafka 集群中消费消息的终端或服务。
  6. consumer group：high-level consumer API 中，每个 consumer 都属于一个 consumer group，每条消息只能被 consumer group 中的一个 Consumer 消费，但可以被多个 consumer group 消费。
  7. replica：partition 的副本，保障 partition 的高可用。
  8. leader：replica 中的一个角色， producer 和 consumer 只跟 leader 交互。
  9. follower：replica 中的一个角色，从 leader 中复制数据。
  10. controller：kafka 集群中的其中一个服务器，用来进行 leader election 以及 各种 failover。
  11. zookeeper：kafka 通过 zookeeper 来存储集群的 meta 信息。

Kafka原理

通常来讲，消息模型可以分为两种：队列和发布-订阅式。队列的处理方式是一组消费者从服务器读取消息，一条消息只有其中的一个消费者来处理。在发布-订阅模型中，消息被广播给所有的消费者，接收到消息的消费者都可以处理此消息。Kafka为这两种模型提供了单一的消费者抽象模型： 消费者组(consumer group)。消费者用一个消费者组名标记自己。

一个发布在Topic上消息被分发给此消费者组中的一个消费者。假如所有的消费者都在一个组中，那么这就变成了queue模型。假如所有的消费者都在不同的组中，那么就完全变成了发布-订阅模型。更通用的， 我们可以创建一些消费者组作为逻辑上的订阅者。每个组包含数目不等的消费者，一个组内多个消费者可以用来扩展性能和容错。

并且，kafka能够保证生产者发送到一个特定的Topic的分区上，消息将会按照它们发送的顺序依次加入，也就是说，如果一个消息M1和M2使用相同的producer发送，M1先发送，那么M1将比M2的offset低，并且优先的出现在日志中。消费者收到的消息也是此顺序。如果一个Topic配置了复制因子（replication facto）为N,那么可以允许N-1服务器宕机而不丢失任何已经提交（committed）的消息。此特性说明kafka有比传统的消息系统更强的顺序保证。但是，相同的消费者组中不能有比分区更多的消费者，否则多出的消费者一直处于空等待，不会收到消息。

Demo

监控一个文件实时采集新增的数据输出到Kafka
FLume采用 exec source + memory channel+ kafka sink

  1. 设置agent

Flume agent配置存储在本地配置文件中。配置文件包含代理中每个source，sink和channel的属性以及它们如何连接在一起以形成数据流

下文$FLUME_HOME，$KAFKA_HOME指FLUME,KAFKA安装目录，

进入$FLUME_HOME/conf 创建exec-memory-kafca.conf并配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 给agent命名
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 配置source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /Users/null/data/flume-test.log
a1.sources.r1.channels = c1

# 配置sink
a1.sinks.k1.type = org.apache.flume.sink.kafca.KafkaSink
a1.sinks.k1.topic = kafka-test
a1.sinks.k1.brokerList = localhost:9092
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 2
a1.sinks.k1.channel = c1
# 配置memory channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

  1. 启动zookeeper
    Kafka需要zookeeper环境

    1
    bin/zkServer.sh start

  2. 启动Kafka

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

  3. 创建话题

    1
    kafka-topics.sh --list --zookeeper localhost:2181 kafka-test

  4. 启动一个Consumer

    1
    kafka-console-consumer.sh --zookeeper localhost:2181 --topic kafka-test --from-beginning

  5. 启动FLume

    1
    flume-ng agent --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/exec-memory-kafca.conf --name a1 -Dflume.root.logger=INFO,console

  6. 在监控的文件中添加内容，可以查看到kafka发出并消费

参考：
1.Kafka基本知识整理
2.FLume架构以及应用