flume简要说明

发表于 | 分类于 | | 阅读次数:
字数统计: 795 | 阅读时长 ≈ 3

Flume简要说明笔记

##　前言

flume是Apache旗下的开源日志收集、传输框架（由Cloudera于2009年捐赠）。可支持收集来自文本、HDFS、HBase等多种数据源的日志记录，并可传输到多种数据源。由于其稳定的性能及表现，目前已被广泛采用。

主体介绍

Event(事件)是Flume的基本传输单元，可理解为传输的数据如日志等。

Agent是Flume的核心运行逻辑。一个Agent就是一个完整的数据收集工具（表现为一个JVM）。包含3个主要组件：Source、Channel、Sink。如下图数据流程模型所示，数据从外部流入Flume（Agent）,分别经过3个组件流转后被送到目的数据源。下面是其介绍：

Source

数据收集端，负责将数据封装成Event,后传输给Channel。

Flume提供了各种source的实现，包括Avro Source、 Exce Source、Spooling Directory Source、 NetCat Source、 Syslog Source、 Syslog TCP Source、Syslog UDP Source、 HTTP Source、 HDFS Source等，下面列举一些常用的Source

Source 说明 必填属性示例
Avro Source 支持Avro协议的数据源 type=avro,bind=127.0.0.1,port=7777
Exce Source 通过运行unix命令收集数据源，如 tail -F等 type=exec,command=xxx
Spooling Directory Source 监听指定目录的文件变化收集日志，这里特别需注意： 一旦文件移到flume指定监视目录里，文件就不能再变化了，否则Flume会失败退出 type=spooldir,spoolDir=path
NetCat Source 监听tcp端口数据 type=netcat,bind=127.0.0.1,port=7777

Channel

数据传输队列，连接Source和Sink,可理解为缓冲区。channel可将数据暂存在内存或是磁盘中，直到Sink处理完将该数据再删除。

channel可分为Memory Channel(内存channel)、FileChannel(文件Channel)等。顾名思义，内存Channel将数据存于内存，FileChannel存于磁盘文件中。还有Spillable Memory Channel，内存满了向磁盘存储，此channel正处于试验性质，不建议生产使用。见Flume Spillable Memory Channel

Sink

从Channel中取出事件，再将其发送出去，是Flume的出口位置。如可将数据发往HDFS、DB、Kafka等。

下列是支持的输出源：

  • DHFS Sink
  • Logger Sink
  • Avro Sink
  • Thrift Sink
  • Null Sink
  • HBase Sink
  • ElasticSearch Sink
  • Custom Sink :自定义sink

Flume数据流程如下所示

Flume配置项

选用exec方式监听日志的方式，收集日志后送到kafka中。配置如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# 配置项，定义源，通道以及sink
agent.sources = mysource
agent.channels = mychannel
agent.sinks = mysink

# 定义数据源
# 数据源类型，unix的二进制命令
agent.sources.mysource.type = exec
# 执行命令
agent.sources.mysource.command = tail -F  /usr/local/log/spark/streaming/spark-stream.log
# 数据源接收通道
agent.sources.mysource.channels = mychannel


# 定义数据通道
# 通道类型，有文件和内存类型，这里定义为内存
agent.channels.mychannel.type = memory
# 在channel中最大存储的容量
agent.channels.mychannel.capacity = 1000
# 从通道中每次事务获取的最大数
agent.channels.mychannel.transactionCapacity = 100

# 定义数据下沉出口
# 定义kafka类型
agent.sinks.mysink.type = org.apache.flume.sink.kafka.KafkaSink
# 要连接的kafka服务器
agent.sinks.mysink.kafka.bootstrap.servers = 127.0.0.1:9092
# kafka 主题
agent.sinks.mysink.kafka.topic = test
# 上游channel 名称
agent.sinks.mysink.channel = mychannel
分享到：
0%