Flume篇---Flume安装配置与相关使用
一.前述
Copy过来一段介绍Apache Flume 是一个从可以收集例如日志，事件等数据资源，并将这些数量庞大的数据从各项数据资源中集中起来存储的工具/服务，或者数集中机制。flume具有高可用，分布式，配置工具，其设计的原理也是基于将数据流，如日志数据从各种网站服务器上汇集起来存储到HDFS，HBase等集中存储器中。官网：http://flume.apache.org/FlumeUserGuide.html
二.架构
1.基本架构
介绍：
Source:（相当于一个来源）
从数据发生器接收数据,并将接收的数据以Flume的event格式传递给一个或者多个通道channal,Flume提供多种数据接收的方式,比如Avro,Thrift,twitter1%等
Channel:（相当于一个中转）
channal是一种短暂的存储容器,它将从source处接收到的event格式的数据缓存起来,直到它们被sinks消费掉,它在source和sink间起着一共桥梁的作用,channal是一个完整的事务,这一点保证了数据在收发的时候的一致性. 并且它可以和任意数量的source和sink链接. 支持的类型有: JDBC channel , File System channel , Memort channel等.
sink:（相当于最后的写出）
sink将数据存储到集中存储器比如Hbase和HDFS,它从channals消费数据(events)并将其传递给目标地. 目标地可能是另一个sink,也可能HDFS,HBase.
2.延伸架构
2.1利用AVRO中转
2.2一般多个来源时可以配置这样
ps:
1、上传
2、解压
3、修改conf/flume-env.sh 文件中的JDK目录
注意：JAVA_OPTS 配置 如果我们传输文件过大 报内存溢出时 需要修改这个配置项
4、验证安装是否成功 ./flume-ng version
5、配置环境变量
export FLUME_HOME=/home/apache-flume-1.6.0-bin
3.2 Source、Channel、Sink有哪些类型
Flume Source
Source类型 | 说明
Avro Source | 支持Avro协议（实际上是Avro RPC），内置支持
Thrift Source | 支持Thrift协议，内置支持
Exec Source | 基于Unix的command在标准输出上生产数据
JMS Source | 从JMS系统（消息、主题）中读取数据
Spooling Directory Source | 监控指定目录内数据变更
Twitter 1% firehose Source| 通过API持续下载Twitter数据，试验性质
Netcat Source | 监控某个端口，将流经端口的每一个文本行数据作为Event输入
Sequence Generator Source | 序列生成器数据源，生产序列数据
Syslog Sources | 读取syslog数据，产生Event，支持UDP和TCP两种协议
HTTP Source | 基于HTTP POST或GET方式的数据源，支持JSON、BLOB表示形式
Legacy Sources | 兼容老的Flume OG中Source（0.9.x版本）
Flume Channel
Channel类型 说明
Memory Channel | Event数据存储在内存中
JDBC Channel | Event数据存储在持久化存储中，当前Flume Channel内置支持Derby
File Channel | Event数据存储在磁盘文件中
Spillable Memory Channel | Event数据存储在内存中和磁盘上，当内存队列满了，会持久化到磁盘文件
Pseudo Transaction Channel | 测试用途
Custom Channel | 自定义Channel实现
Flume Sink
Sink类型 说明
HDFS Sink | 数据写入HDFS
Logger Sink | 数据写入日志文件
Avro Sink | 数据被转换成Avro Event，然后发送到配置的RPC端口上
Thrift Sink | 数据被转换成Thrift Event，然后发送到配置的RPC端口上
IRC Sink | 数据在IRC上进行回放
File Roll Sink | 存储数据到本地文件系统
Null Sink | 丢弃到所有数据
HBase Sink | 数据写入HBase数据库
Morphline Solr Sink | 数据发送到Solr搜索服务器（集群）
ElasticSearch Sink | 数据发送到Elastic Search搜索服务器（集群）
Kite Dataset Sink | 写数据到Kite Dataset，试验性质的
Custom Sink | 自定义Sink实现
案例1、 A simple example
http://flume.apache.org/FlumeUserGuide.html#a-simple-example
配置文件
############################################################
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
############################################################
启动flume
flume-ng agent -n a1 -c conf -f simple.conf -Dflume.root.logger=INFO,console 指定配置目录
安装telnet
yum install telnet
退出 ctrl+] quit
Memory Chanel 配置
capacity：默认该通道中最大的可以存储的event数量是100，
trasactionCapacity：每次最大可以source中拿到或者送到sink中的event数量也是100
keep-alive：event添加到通道中或者移出的允许时间
byte**：即event的字节量的限制，只包括eventbody
案例2、两个flume做集群（第一个agent的sink作为第二个agent的source）
node01服务器中，配置文件
############################################################
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = node1
a1.sources.r1.port = 44444
# Describe the sink
# a1.sinks.k1.type = logger
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = node2
a1.sinks.k1.port = 60000
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
############################################################
node02服务器中，安装Flume（步骤略）
配置文件
############################################################
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.bind = node2
a1.sources.r1.port = 60000
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
############################################################
先启动node02的Flume
flume-ng agent -n a1 -c conf -f avro.conf -Dflume.root.logger=INFO,console
再启动node01的Flume
flume-ng agent -n a1 -c conf -f simple.conf2 -Dflume.root.logger=INFO,console
打开telnet 测试 node02控制台输出结果
案例3、Exec Source（监听一个文件）
http://flume.apache.org/FlumeUserGuide.html#exec-source
配置文件
############################################################
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/flume.exec.log
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
############################################################
启动Flume
flume-ng agent -n a1 -c conf -f exec.conf -Dflume.root.logger=INFO,console
创建空文件演示 touch flume.exec.log
循环添加数据
for i in {1..50}; do echo "$i hi flume" >> flume.exec.log ; sleep 0.1; done
案例4、Spooling Directory Source（监听一个目录）
http://flume.apache.org/FlumeUserGuide.html#spooling-directory-source
配置文件
############################################################
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /home/logs
a1.sources.r1.fileHeader = true
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
############################################################
启动Flume
flume-ng agent -n a1 -c conf -f spool.conf -Dflume.root.logger=INFO,console
拷贝文件演示
mkdir logs
cp flume.exec.log logs/
案例5、hdfs sink
http://flume.apache.org/FlumeUserGuide.html#hdfs-sink
配置文件
############################################################
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /home/logs
a1.sources.r1.fileHeader = true
# Describe the sink
***只修改上一个spool sink的配置代码块 a1.sinks.k1.type = logger
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path=hdfs://sxt/flume/%Y-%m-%d/%H%M
##每隔60s或者文件大小超过10M的时候产生新文件
# hdfs有多少条消息时新建文件，0不基于消息个数
a1.sinks.k1.hdfs.rollCount=0
# hdfs创建多长时间新建文件，0不基于时间
a1.sinks.k1.hdfs.rollInterval=60
# hdfs多大时新建文件，0不基于文件大小
a1.sinks.k1.hdfs.rollSize=10240
# 当目前被打开的临时文件在该参数指定的时间（秒）内，没有任何数据写入，则将该临时文件关闭并重命名成目标文件
a1.sinks.k1.hdfs.idleTimeout=3
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.useLocalTimeStamp=true
## 每五分钟生成一个目录:
# 是否启用时间上的”舍弃”，这里的”舍弃”，类似于”四舍五入”，后面再介绍。如果启用，则会影响除了%t的其他所有时间表达式
a1.sinks.k1.hdfs.round=true
# 时间上进行“舍弃”的值；
a1.sinks.k1.hdfs.roundValue=5
# 时间上进行”舍弃”的单位，包含：second,minute,hour
a1.sinks.k1.hdfs.roundUnit=minute
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1（将source,channel,sink关联）
############################################################
创建HDFS目录
hadoop fs -mkdir /flume
启动Flume
flume-ng agent -n a1 -c conf -f hdfs.conf -Dflume.root.logger=INFO,console
查看hdfs文件
hadoop fs -ls /flume/...
hadoop fs -get /flume/...
http://flume.apache.org/
安装
1、上传
2、解压
3、修改conf/flume-env.sh 文件中的JDK目录
注意：JAVA_OPTS 配置 如果我们传输文件过大 报内存溢出时 需要修改这个配置项
4、验证安装是否成功 ./flume-ng version
5、配置环境变量
export FLUME_HOME=/home/apache-flume-1.6.0-bin
Source、Channel、Sink有哪些类型
Flume Source
Source类型 | 说明
Avro Source | 支持Avro协议（实际上是Avro RPC），内置支持
Thrift Source | 支持Thrift协议，内置支持
Exec Source | 基于Unix的command在标准输出上生产数据
JMS Source | 从JMS系统（消息、主题）中读取数据
Spooling Directory Source | 监控指定目录内数据变更
Twitter 1% firehose Source| 通过API持续下载Twitter数据，试验性质
Netcat Source | 监控某个端口，将流经端口的每一个文本行数据作为Event输入
Sequence Generator Source | 序列生成器数据源，生产序列数据
Syslog Sources | 读取syslog数据，产生Event，支持UDP和TCP两种协议
HTTP Source | 基于HTTP POST或GET方式的数据源，支持JSON、BLOB表示形式
Legacy Sources | 兼容老的Flume OG中Source（0.9.x版本）
Flume Channel
Channel类型 说明
Memory Channel | Event数据存储在内存中
JDBC Channel | Event数据存储在持久化存储中，当前Flume Channel内置支持Derby
File Channel | Event数据存储在磁盘文件中
Spillable Memory Channel | Event数据存储在内存中和磁盘上，当内存队列满了，会持久化到磁盘文件
Pseudo Transaction Channel | 测试用途
Custom Channel | 自定义Channel实现
Flume Sink
Sink类型 说明
HDFS Sink | 数据写入HDFS
Logger Sink | 数据写入日志文件
Avro Sink | 数据被转换成Avro Event，然后发送到配置的RPC端口上
Thrift Sink | 数据被转换成Thrift Event，然后发送到配置的RPC端口上
IRC Sink | 数据在IRC上进行回放
File Roll Sink | 存储数据到本地文件系统
Null Sink | 丢弃到所有数据
HBase Sink | 数据写入HBase数据库
Morphline Solr Sink | 数据发送到Solr搜索服务器（集群）
ElasticSearch Sink | 数据发送到Elastic Search搜索服务器（集群）
Kite Dataset Sink | 写数据到Kite Dataset，试验性质的
Custom Sink | 自定义Sink实现
案例1、 A simple example
http://flume.apache.org/FlumeUserGuide.html#a-simple-example
配置文件
############################################################
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
############################################################
启动flume
flume-ng agent -n a1 -c conf -f simple.conf -Dflume.root.logger=INFO,console
安装telnet
yum install telnet
退出 ctrl+] quit
Memory Chanel 配置
capacity：默认该通道中最大的可以存储的event数量是100，
trasactionCapacity：每次最大可以source中拿到或者送到sink中的event数量也是100
keep-alive：event添加到通道中或者移出的允许时间
byte**：即event的字节量的限制，只包括eventbody
案例2、两个flume做集群
node01服务器中，配置文件
############################################################
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = node1
a1.sources.r1.port = 44444
# Describe the sink
# a1.sinks.k1.type = logger
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = node2
a1.sinks.k1.port = 60000
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
############################################################
node02服务器中，安装Flume（步骤略）
配置文件
############################################################
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.bind = node2
a1.sources.r1.port = 60000
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
############################################################
先启动node02的Flume
flume-ng agent -n a1 -c conf -f avro.conf -Dflume.root.logger=INFO,console
再启动node01的Flume
flume-ng agent -n a1 -c conf -f simple.conf2 -Dflume.root.logger=INFO,console
打开telnet 测试 node02控制台输出结果
案例3、Exec Source
http://flume.apache.org/FlumeUserGuide.html#exec-source
配置文件
############################################################
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/flume.exec.log
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
############################################################
启动Flume
flume-ng agent -n a1 -c conf -f exec.conf -Dflume.root.logger=INFO,console
创建空文件演示 touch flume.exec.log
循环添加数据
for i in {1..50}; do echo "$i hi flume" >> flume.exec.log ; sleep 0.1; done
案例4、Spooling Directory Source
http://flume.apache.org/FlumeUserGuide.html#spooling-directory-source
配置文件
############################################################
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /home/logs
a1.sources.r1.fileHeader = true
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
############################################################
启动Flume
flume-ng agent -n a1 -c conf -f spool.conf -Dflume.root.logger=INFO,console
拷贝文件演示
mkdir logs
cp flume.exec.log logs/
案例5、hdfs sink
http://flume.apache.org/FlumeUserGuide.html#hdfs-sink
配置文件
############################################################
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /home/logs
a1.sources.r1.fileHeader = true
# Describe the sink
***只修改上一个spool sink的配置代码块 a1.sinks.k1.type = logger
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path=hdfs://sxt/flume/%Y-%m-%d/%H%M
##每隔60s或者文件大小超过10M的时候产生新文件
# hdfs有多少条消息时新建文件，0不基于消息个数
a1.sinks.k1.hdfs.rollCount=0
# hdfs创建多长时间新建文件，0不基于时间
a1.sinks.k1.hdfs.rollInterval=60
# hdfs多大时新建文件，0不基于文件大小
a1.sinks.k1.hdfs.rollSize=10240
# 当目前被打开的临时文件在该参数指定的时间（秒）内，没有任何数据写入，则将该临时文件关闭并重命名成目标文件
a1.sinks.k1.hdfs.idleTimeout=3
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.useLocalTimeStamp=true
## 每五分钟生成一个目录:
# 是否启用时间上的”舍弃”，这里的”舍弃”，类似于”四舍五入”，后面再介绍。如果启用，则会影响除了%t的其他所有时间表达式
a1.sinks.k1.hdfs.round=true
# 时间上进行“舍弃”的值；
a1.sinks.k1.hdfs.roundValue=5
# 时间上进行”舍弃”的单位，包含：second,minute,hour
a1.sinks.k1.hdfs.roundUnit=minute
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
############################################################
创建HDFS目录
hadoop fs -mkdir /flume
启动Flume
flume-ng agent -n a1 -c conf -f hdfs.conf -Dflume.root.logger=INFO,console
查看hdfs文件
hadoop fs -ls /flume/...
hadoop fs -get /flume/...
作业：
1、flume如何收集java请求数据
2、项目当中如何来做？ 日志存放/log/目录下 以yyyyMMdd为子目录 分别存放每天的数据
转载于:https://www.cnblogs.com/LHWorldBlog/p/8305177.html
Flume篇---Flume安装配置与相关使用相关推荐
- Tomcat的安装配置及相关问题解决
Apache Tomcat/8.5.82 小声说 Tomcat的安装配置及相关问题解决 Apache Tomcat/8.5.82的下载 配置 访问Tomcat 常出现的问题及解决 1.tomcat8. ...
- Java SE 7 Update 17的安装配置及相关问题解决
Java SE 7 Update 17 小声说 Eclipse Java EE集成开发环境 下载 安装 配置 总结 小声说 首先非常感谢大家的认可,近一周来收到些私信,有些没及时回复十分不好意思,但是 ...
- phpmyadmin安装配置以及相关问题
phpmyadmin配置文件: 配置提示:几乎所有的配置参数都在 config.inc.php 文件中. 如果这个文件不存在,您可以在 libraries 目录中找到 config.default.p ...
- linux 环境安装配置,Linux相关环境安装与配置
最近买了一个腾讯云的服务器,简单安装了下环境并记录一下 Java: 1.查找java相关的列表 yum -y list java* 2.安装JDK(这里是1.8.0版本) yum install ja ...
- gradle教程 [原创]（eclipse/ADT下 非插件 非Android Studio/AS）纯手打 第一篇：安装配置gradle...
一个bug 一个脚印的叫你们用gradle. 1介于网络上的很多资料都是老的 不适用与现在的新版本gradle 尤其是有些gradle方法改名了老的用不了 2介于网上都是粘贴复制并且零碎我很蛋疼啊,走 ...
- [ 常用工具篇 ] burpsuite_pro 安装配置详解（附安装包）
- Hadoop、Slurm平台详细安装配置步骤
大数据Hadoop.Slurm平台 安 装 配 置 手 册 河北科技大学理学院 王子元 2018年6月 绪论 本人本科毕业设计选的是<数学系大数据环境建设>课题,实验集群采用四台惠普工作站 ...
- airflow mysql_AirFlow 安装配置
airflow 安装配置 airflow 相关软件安装 python 3.6.5 安装 安装依赖程序 : [root@node01 ~]# yum -y install zlib zlib-devel ...
- 【原创】flume-1.3.0安装配置以及flume启动说明
Flume介绍 一 flume基本介绍 Apache flume是一个分布式的.可靠的和可用的用来高效收集.同济和移动大量数据从众多不同sources到一个集中的数据存储库的系统. Flume eve ...
最新文章
- YOLO v1到YOLO v4（下）
- 教程 | Caffe在Windows10系统上安装与配置
- python批量重命名指定目录下所有文件的后缀名
- android 文件下载 超简单
- 关于css选择器的问题 + * ~这三个有什么区别
- 上周热点回顾（12.18-12.24）
- V-rep学习笔记：机器人逆运动学解算
- python赋值语句格式_Python赋值语句后逗号的作用分析
- apm固定翼调试方法
- WCF和webservice的区别
- oracle获取去年年份_「实战」中文检错纠错之语料获取与处理
- viewer 获取当前显示的图片信息_通过这个 Runtime 统计信息可视化库学到了什么？...
- 2018蓝桥模拟赛（一） 数独（dfs）
- HDU 4069 Squiggly Sudoku
- 哈夫曼编码（Huffman Coding）原理详解
- springboot发送垃圾邮件
- SAP ERP和ORACLE ERP的区别是哪些？
- java工资高还是web前端工资高_java工资高还是php,java工资高还是web前端工资高,php和java学费...
- dva是什么游戏_守望先锋：DVA这个皮肤小蛮腰不算什么，全部细节是这个小脚丫...
- win10家庭版 mysql_MySQL下载安装详解（win10家庭版）
热门文章
- 公有链、私有链、联盟链、许可链，这些区块链又分别代表着什么意思？
- 忘记commit也会造成select查询的性能问题
- windows server 2012/2016 开启桌面图标
- java处理json的工具类(list,map和json的之间的转换)
- Android自定义view之ViewPager指示器——1
- jenkins手把手教你从入门到放弃03-安装Jenkins时web界面出现该jenkins实例似乎已离线
- python 删除文件 通配符_python 实现删除文件或文件夹实例详解
- java元类_元类 - 一心不乱 - 博客园
- 获取上周_北美木材价格上周趋于稳定
- 本地算术验证码识别教程