设为首页 加入收藏

TOP

Flume + Kafka学习——整合Demo
2019-04-14 01:49:21 】 浏览:19
Tags:Flume Kafka 学习 整合 Demo

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并配置

Exec Source配置

type:

The component type name, needs to be exec

command:

The command to execute

Kafka Sink 配置

type:

Must be set to org.apache.flume.sink.kafka.KafkaSink

brokerList:

List of brokers Kafka-Sink will connect to, to get the list of topic partitions This can be a partial list of brokers, but we recommend at least two for HA. The format is comma separated list of hostname:port

topic:

default-flume-topic    The topic in Kafka to which the messages will be published. If this parameter is configured, messages will be published to this topic. If the event header contains a “topic” field, the event will be published to that topic overriding the topic configured here.

batchSize:

How many messages to process in one batch. Larger batches improve throughput while adding latency.

requiredAcks:

How many replicas must acknowledge a message before its considered successfully written. Accepted values are 0 (Never wait for acknowledgement), 1 (wait for leader only), -1 (wait for all replicas) Set this to -1 to avoid data loss in some cases of leader failure.

详细配置见官方文档

# 给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.kafka.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环境

    bin/zkServer.sh start
  2. 启动Kafka

    bin/kafka-server-start.sh config/server.properties
  3. 创建话题

    kafka-test 为topic 名称与flume 配置中a1.sinks.k1.topic 对应

    kafka-topics.sh --list --zookeeper localhost:2181 kafka-test
  4. 启动一个Consumer

    kafka-console-consumer.sh --zookeeper localhost:2181 --topic kafka-test --from-beginning
    
  5. 启动FLume

    --conf 指定配置文件所在位置 --conf-file为指定配置文件 --name 为agent的名称

    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架构以及应用


编程开发网
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇ICC副本>>>>(logback.. 下一篇flume读取日志数据写入kafka &nbs..

评论

帐  号: 密码: (新用户注册)
验 证 码:
表  情:
内  容:

array(4) { ["type"]=> int(8) ["message"]=> string(24) "Undefined variable: jobs" ["file"]=> string(32) "/mnt/wp/cppentry/do/bencandy.php" ["line"]=> int(217) }