Flume介绍安装使用

阅读原文时间：2023年07月09日阅读：3

APache Flume官网：http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html#memory-channel

Flume的介绍
Flume的安装

Flume的介绍

Flume简介

flume是一个分布式、可靠、和高可用的海量日志采集、聚合和传输的系统。

flume的数据流由事件(Event)贯穿始终。事件是Flume的基本数据单位，它携带日志数据(字节数组形式)并且携带有头信息，这些Event由Agent外部的Source生成，当Source捕获事件后会进行特定的格式化，然后Source会把事件推入(单个或多个)Channel中。你可以把Channel看作是一个缓冲区，它将保存事件直到Sink处理完该事件。Sink负责持久化日志或者把事件推向另一个Source。

Flume架构

Flume 运行的核心是 Agent。Flume以agent为最小的独立运行单位。一个agent就是一个JVM。

Flume数据流

Flume 的核心是把数据从数据源收集过来，再送到目的地。为了保证输送一定成功，在送到目的地之前，会先缓存数据，待数据真正到达目的地后，删除自己缓存的数据
Flume 传输的数据的基本单位是 Event，如果是文本文件，通常是一行记录，这也是事务的基本单位。 Event 从 Source，流向 Channel，再到 Sink，本身为一个 byte 数组，并可携带 headers 信息。 Event 代表着一个数据流的最小完整单元，从外部数据源来，向外部的目的地去。
值得注意的是，Flume提供了大量内置的Source、Channel和Sink类型。不同类型的Source,Channel和Sink可以自由组合。组合方式基于用户设置的配置文件，非常灵活。
比如：Channel可以把事件暂存在内存里，也可以持久化到本地硬盘上。Sink可以把日志写入HDFS, HBase，甚至是另外一个Source等等。Flume支持用户建立多级流，也就是说，多个agent可以协同工作。

Flume可靠性

Flume 使用事务性的方式保证传送Event整个过程的可靠性。 Sink 必须在Event 已经被传达到下一站agent里，又或者，已经被存入外部数据目的地之后，才能把 Event 从 Channel 中 remove 掉。这样数据流里的 event 无论是在一个 agent 里还是多个 agent 之间流转，都能保证可靠，因为以上的事务保证了 event 会被成功存储起来。比如 Flume支持在本地保存一份channel文件作为备份，而memory channel 将event存在内存 queue 里，速度快，但丢失的话无法恢复。

多个Agent的数据汇聚到同一个Agent

这种情况应用的场景比较多，比如要收集Web网站的用户行为日志， Web网站为了可用性使用的负载集群模式，每个节点都产生用户行为日志，可以为每个节点都配置一个Agent来单独收集日志数据，然后多个Agent将数据最终汇聚到一个用来存储数据存储系统，如HDFS上。

Flume的安装

1、上传至虚拟机，并解压

tar -zxvf apache-flume-1.9.0-bin.tar.gz -C /usr/local/soft/

在环境变量中增加如下命令，可以使用 soft 快速切换到 /usr/local/soft

alias soft='cd /usr/local/soft/'

2、重命名目录，并配置环境变量

mv apache-flume-1.9.0-bin/ flume-1.9.0
vim /etc/profile
source /etc/profile

3、查看flume版本

flume-ng version


[root@master soft]# flume-ng version
Flume 1.9.0
Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git
Revision: d4fcab4f501d41597bc616921329a4339f73585e
Compiled by fszabo on Mon Dec 17 20:45:25 CET 2018
From source with checksum 35db629a3bda49d23e9b3690c80737f9
[root@master soft]#

4、测试flume

监控一个目录，将数据打印出来
新建目录
```
mkdir /usr/local/data/flume/
```
- 配置文件
  
  首先先给agent起一个名字叫a1
  
  分别给source channel sink取名字
  
  a1.sources = r1
  a1.channels = c1
  a1.sinks = k1
  
  分别对source、channel、sink进行配置
  
  配置source
  
  将source的类型指定为 spooldir 用于监听一个目录下文件的变化
  
  因为每个组件可能会出现相同的属性名称，所以在对每个组件进行配置的时候
  
  需要加上 agent的名字.sources.组件的名字.属性 = 属性值
  
  fileSuffix:更改文件后缀名
  
  fileHeader:是否添加存储绝对路径文件名的标头
  
  a1.sources.r1.type = spooldir
  a1.sources.r1.spoolDir = /root/data/
  a1.sources.r1.fileSuffix = .zyl
  a1.sources.r1.fileHeader = true
  
  给r1这个souces配置一个拦截器并取名为 i1
  
  a1.sources.r1.interceptors = i1
  
  将拦截器i1的类型设置为timestamp 会将处理数据的时间以毫秒的格式插入event的header中
  
  a1.sources.r1.interceptors.i1.type = timestamp
  
  将拦截器i1的类型设置为regex_filter 会根据正则表达式过滤数据
  
  a1.sources.r1.interceptors.i1.type = regex_filter
  
  配置正则表达式
  
  a1.sources.r1.interceptors.i1.regex = \d{3,6}
  
  excludeEvents = true 表示将匹配到的过滤，未匹配到的放行
  
  a1.sources.r1.interceptors.i1.excludeEvents = true
  
  配置sink
  
  使用logger作为sink组件，可以将收集到数据直接打印到控制台
  
  a1.sinks.k1.type = logger
  
  配置channel
  
  将channel的类型设置为memory，表示将event缓存在内存中
  
  a1.channels.c1.type = memory
  
  组装
  
  将sources的channels属性指定为c1
  
  a1.sources.r1.channels = c1
  
  将sinks的channel属性指定为c1
  
  a1.sinks.k1.channel = c1
- 启动agent
  
  flume-ng agent -n a1 -f ./flumetest -Dflume.root.logger=DEBUG,console
- 在/usr/local/data/flume/目录下新建文件，输入内容，观察flume进程打印的日志
  
  随意在a.txt中加入一些内容
  
  vim 1

5、flume的使用

spoolingToHDFS.conf
- 配置文件
  
  a表示给agent命名为a
  
  给source组件命名为r1
  
  a.sources = r1
  
  给sink组件命名为k1
  
  a.sinks = k1
  
  给channel组件命名为c1
  
  a.channels = c1
  #指定spooldir的属性
  a.sources.r1.type = spooldir
  a.sources.r1.spoolDir = /usr/local/data/flume/
  a.sources.r1.fileHeader = true
  a.sources.r1.interceptors = i1
  a.sources.r1.interceptors.i1.type = timestamp
  #指定sink的类型
  a.sinks.k1.type = hdfs
  a.sinks.k1.hdfs.path = /flume/data/dir1
  
  指定文件名前缀
  
  a.sinks.k1.hdfs.filePrefix = student
  
  指定达到多少数据量写一次文件单位：bytes
  
  a.sinks.k1.hdfs.rollSaize = 102400
  
  指定多少条写一次文件
  
  a.sinks.k1.hdfs.rollCount = 1000
  
  指定文件类型为流来什么输出什么
  
  a.sinks.k1.hdfs.fileType = DataStream
  
  指定文件输出格式为text
  
  a.sinks.k1.hdfs.writeFormat = text
  
  指定文件名后缀
  
  a.sinks.k1.hdfs.fileSuffix = .txt
  
  #指定channel
  a.channels.c1.type = memory
  #存储在通道中的最大事件数
  a.channels.c1.capacity = 1000
  
  表示sink每次会从channel里取多少数据
  
  a.channels.c1.transactionCapacity = 100
  
  组装
  
  a.sources.r1.channels = c1
  a.sinks.k1.channel = c1
- 在 /usr/local/data/flume/目录下准备数据
  
  The Zen of Python, by Tim Peters
  
  Beautiful is better than ugly.
  Explicit is better than implicit.
  Simple is better than complex.
  Complex is better than complicated.
  Flat is better than nested.
  Sparse is better than dense.
  Readability counts.
  Special cases aren't special enough to break the rules.
  Although practicality beats purity.
  Errors should never pass silently.
  Unless explicitly silenced.
  In the face of ambiguity, refuse the temptation to guess.
  There should be one-- and preferably only one --obvious way to do it.
  Although that way may not be obvious at first unless you're Dutch.
  Now is better than never.
  Although never is often better than right now.
  If the implementation is hard to explain, it's a bad idea.
  If the implementation is easy to explain, it may be a good idea.
  Namespaces are one honking great idea -- let's do more of those!
- 启动agent
  
  flume-ng agent -n a -f ./spoolingToHDFS.conf -Dflume.root.logger=DEBUG,console

hbaseLogToHDFS
- 配置文件
  
  a表示给agent命名为a
  
  给source组件命名为r1
  
  a.sources = r1
  
  给sink组件命名为k1
  
  a.sinks = k1
  
  给channel组件命名为c1
  
  a.channels = c1
  #指定exec的属性
  a.sources.r1.type = exec
  a.sources.r1.command = tail -f /usr/local/soft/hbase-1.4.6/logs/hbase-root-master-master.log
  #指定sink的类型
  a.sinks.k1.type = hdfs
  a.sinks.k1.hdfs.path = /flume/data/dir2
  
  指定文件名前缀
  
  a.sinks.k1.hdfs.filePrefix = hbaselog
  
  指定达到多少数据量写一次文件单位：bytes
  
  a.sinks.k1.hdfs.rollSize = 102400
  
  指定多少条写一次文件
  
  a.sinks.k1.hdfs.rollCount = 1000
  
  指定文件类型为流来什么输出什么
  
  a.sinks.k1.hdfs.fileType = DataStream
  
  指定文件输出格式为text
  
  a.sinks.k1.hdfs.writeFormat = text
  
  指定文件名后缀
  
  a.sinks.k1.hdfs.fileSuffix = .txt
  
  #指定channel
  a.channels.c1.type = memory
  a.channels.c1.capacity = 1000
  
  表示sink每次会从channel里取多少数据
  
  a.channels.c1.transactionCapacity = 100
  
  组装
  
  a.sources.r1.channels = c1
  a.sinks.k1.channel = c1z
hbaselogToHBase
- 在hbase中创建log表
  
  create 'log','cf1'
- 配置文件
  
  a表示给agent命名为a
  
  给source组件命名为r1
  
  a.sources = r1
  
  给sink组件命名为k1
  
  a.sinks = k1
  
  给channel组件命名为c1
  
  a.channels = c1
  #指定exec的属性
  a.sources.r1.type = exec
  a.sources.r1.command = cat /usr/local/soft/hbase-1.4.6/logs/hbase-root-master-master.log
  #指定sink的类型
  a.sinks.k1.type = hbase
  a.sinks.k1.table = log
  a.sinks.k1.columnFamily = cf1
  
  #指定channel
  a.channels.c1.type = memory
  a.channels.c1.capacity = 100000
  
  表示sink每次会从channel里取多少数据
  
  a.channels.c1.transactionCapacity = 100
  
  组装
  
  a.sources.r1.channels = c1
  a.sinks.k1.channel = c1
netcatLogger（TCP）

监听telnet端口
- 安装telnet
  
  yum install telnet
- 配置文件
  
  a表示给agent命名为a
  
  给source组件命名为r1
  
  a.sources = r1
  
  给sink组件命名为k1
  
  a.sinks = k1
  
  给channel组件命名为c1
  
  a.channels = c1
  #指定netcat的属性
  a.sources.r1.type = netcat
  #0.0.0.0表示所有地址
  a.sources.r1.bind = 0.0.0.0
  a.sources.r1.port = 8888
  
  #指定sink的类型
  a.sinks.k1.type = logger
  #指定channel
  a.channels.c1.type = memory
  a.channels.c1.capacity = 1000
  
  表示sink每次会从channel里取多少数据
  
  a.channels.c1.transactionCapacity = 100
  
  组装
  
  a.sources.r1.channels = c1
  a.sinks.k1.channel = c1
- 启动
  - 先启动agent
    
    flume-ng agent -n a -f ./netcatToLogger.conf -Dflume.root.logger=DEBUG,console
  - 在启动telnet，测试主机的连通性
    
    telnet master 8888

httpToLogger（Http）
- 配置文件
  
  a表示给agent命名为a
  
  给source组件命名为r1
  
  a.sources = r1
  
  给sink组件命名为k1
  
  a.sinks = k1
  
  给channel组件命名为c1
  
  a.channels = c1
  #指定http的属性
  a.sources.r1.type = http
  a.sources.r1.port = 6666
  
  #指定sink的类型
  a.sinks.k1.type = logger
  #指定channel
  a.channels.c1.type = memory
  a.channels.c1.capacity = 1000
  
  表示sink每次会从channel里取多少数据
  
  a.channels.c1.transactionCapacity = 100
  
  组装
  
  a.sources.r1.channels = c1
  a.sinks.k1.channel = c1
- 启动
  - 先启动agent
    
    flume-ng agent -n a -f ./httpToLogger.conf -Dflume.root.logger=DEBUG,console
  - 再使用curl发起一个http请求
    
    curl -X POST -d '[{ "headers" :{"a" : "a1","b" : "b1"},"body" : "hello~http~flume~"},{"headers" :{"a2" : "a11","b2" : "b12"},"body" : "hello2~http2~flume2~"}]' http://master:6666

手机扫一扫

移动阅读更方便

你可能感兴趣的文章

日志采集工具Flume的安装与使用方法

Flume对接Kafka

大数据学习day36-----flume02--------1.avro source和kafka source 2. 拦截器（Interceptor） 3. channel详解 4 sink 5 slector（选择器）6 sink processor

flume集成hdfs(hdfs开启kerberos认证)

Flume+kakfa+sparkStream实时处理数据测试

大数据学习day35----flume01-------1 agent（关于agent的一些问题），2 event，3 有关agent和event的一些问题，4 transaction(事务控制机制)，5 flume安装 6.Flume入门案例

Flume 组件安装配置

Flume介绍安装使用

Flume的介绍

Flume简介

Flume架构

Flume数据流

Flume可靠性

多个Agent的数据汇聚到同一个Agent

Flume的安装

1、上传至虚拟机，并解压

2、重命名目录，并配置环境变量

3、查看flume版本

4、测试flume

首先先给agent起一个名字 叫a1

分别给source channel sink取名字

分别对source、channel、sink进行配置

配置source

将source的类型指定为 spooldir 用于监听一个目录下文件的变化

因为每个组件可能会出现相同的属性名称，所以在对每个组件进行配置的时候

需要加上 agent的名字.sources.组件的名字.属性 = 属性值

fileSuffix:更改文件后缀名

fileHeader:是否添加存储绝对路径文件名的标头

给r1这个souces配置一个拦截器并取名为 i1

将拦截器i1的类型设置为timestamp 会将处理数据的时间以毫秒的格式插入event的header中

a1.sources.r1.interceptors.i1.type = timestamp

将拦截器i1的类型设置为regex_filter 会根据正则表达式过滤数据

配置正则表达式

excludeEvents = true 表示将匹配到的过滤，未匹配到的放行

配置sink

使用logger作为sink组件，可以将收集到数据直接打印到控制台

配置channel

将channel的类型设置为memory，表示将event缓存在内存中

组装

将sources的channels属性指定为c1

将sinks的channel属性指定为c1

随意在a.txt中加入一些内容

5、flume的使用

a表示给agent命名为a

给source组件命名为r1

给sink组件命名为k1

给channel组件命名为c1

指定文件名前缀

指定达到多少数据量写一次文件 单位：bytes

指定多少条写一次文件

指定文件类型为 流 来什么输出什么

指定文件输出格式 为text

指定文件名后缀

表示sink每次会从channel里取多少数据

组装

a表示给agent命名为a

给source组件命名为r1

给sink组件命名为k1

给channel组件命名为c1

指定文件名前缀

指定达到多少数据量写一次文件 单位：bytes

指定多少条写一次文件

指定文件类型为 流 来什么输出什么

指定文件输出格式 为text

指定文件名后缀

表示sink每次会从channel里取多少数据

组装

a表示给agent命名为a

给source组件命名为r1

给sink组件命名为k1

给channel组件命名为c1

表示sink每次会从channel里取多少数据

组装

a表示给agent命名为a

给source组件命名为r1

给sink组件命名为k1

给channel组件命名为c1

表示sink每次会从channel里取多少数据

组装

a表示给agent命名为a

给source组件命名为r1

给sink组件命名为k1

给channel组件命名为c1

表示sink每次会从channel里取多少数据

组装

首先先给agent起一个名字叫a1

指定达到多少数据量写一次文件单位：bytes

指定文件类型为流来什么输出什么

指定文件输出格式为text

指定达到多少数据量写一次文件单位：bytes

指定文件类型为流来什么输出什么

指定文件输出格式为text