当前位置： 首页 > article >正文

Flume的安装及使用

article 2024/10/26 21:48:53

Flume的安装及使用

Flume的安装及使用Flume的安装1、上传至虚拟机，并解压2、重命名目录，并配置环境变量3、查看flume版本4、测试flume5、flume的使用6、多sink7、多Agent

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
export FLUME_HOME=/usr/local/soft/flume-1.9.0
export PATH=$PATH:$FLUME_HOME/bin
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

  • 监控一个目录，将数据打印出来

    • 配置文件

    # 首先先给agent起一个名字 叫a1
# 分别给source channel sink取名字
a1.sources = r1
a1.channels = c1
a1.sinks = k1
​
# 分别对source、channel、sink进行配置
​
# 配置source
# 将source的类型指定为 spooldir 用于监听一个目录下文件的变化
# 因为每个组件可能会出现相同的属性名称，所以在对每个组件进行配置的时候 
# 需要加上 agent的名字.sources.组件的名字.属性 = 属性值
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /root/data/
a1.sources.r1.fileSuffix = .ok
a1.sources.r1.fileHeader = true
​
# 配置channel
# 将channel的类型设置为memory，表示将event缓存在内存中
a1.channels.c1.type = memory
​
# 配置sink
# 使用logger作为sink组件，可以将收集到数据直接打印到控制台
a1.sinks.k1.type = logger
​
# 组装
# 将sources的channels属性指定为c1
a1.sources.r1.channels = c1
​
# 将sinks的channel属性指定为c1
a1.sinks.k1.channel = c1

    • 启动agent

    flume-ng agent -n a1 -f ./spoolingtest.conf -Dflume.root.logger=DEBUG,console

    • 新建/root/data目录

    mkdir /root/data

    • 在/root/data/目录下新建文件，输入内容，观察flume进程打印的日志

    # 随意在a.txt中加入一些内容
vim /root/data/a.txt
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 = /root/data 
a.sources.r1.fileHeader = true 
​
#指定sink的类型
a.sinks.k1.type = hdfs
a.sinks.k1.hdfs.path = /data/flume/students
# 指定文件名前缀
a.sinks.k1.hdfs.filePrefix = student
# 指定达到多少数据量写一次文件 单位：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 = c1

    • 在 /root/data/目录下准备数据

    1500100003,单乐蕊,22,女,理科六班
1500100004,葛德曜,24,男,理科三班
1500100005,宣谷芹,22,女,理科五班
1500100006,边昂雄,21,男,理科二班
1500100007,尚孤风,23,女,文科六班
1500100008,符半双,22,女,理科六班
1500100009,沈德昌,21,男,理科一班
1500100010,羿彦昌,23,男,理科六班
1500100011,宰运华,21,男,理科三班
1500100012,梁易槐,21,女,理科一班
1500100013,逯君昊,24,男,文科二班
1500100014,羿旭炎,23,男,理科五班
1500100015,宦怀绿,21,女,理科一班
1500100016,潘访烟,23,女,文科一班
1500100017,高芷天,21,女,理科五班

    • 启动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 -200f /usr/local/soft/hbase-2.2.7/logs/hbase-root-master-master.log
​
#指定sink的类型
a.sinks.k1.type = hdfs
a.sinks.k1.hdfs.path = /data/flume/hbase_log
# 指定文件名前缀
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 = c1

  • 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 = tail -200f /usr/local/soft/hbase-2.2.7/logs/hbase-root-master-master.log
​
#指定sink的类型
a.sinks.k1.type = hbase2
a.sinks.k1.table = hbase_log
a.sinks.k1.columnFamily = info
​
#指定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
    # 创建hbase表
create 'hbase_log','info'

  • netcatLogger

    监听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 
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

    • 配置文件

    # 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" : "数加学院"}]' http://master:6666
```
6、多sink

vim netcat_to_hdfs.conf

# a表示给agent命名为a
# 给source组件命名为r1
a.sources = r1
# 给sink组件命名为k1
a.sinks = k1 k2
# 给channel组件命名为c1
a.channels = c1
​
#指定netcat的属性
a.sources.r1.type = netcat 
a.sources.r1.bind = 0.0.0.0 
a.sources.r1.port = 8888 
​
#指定sink的类型
a.sinks.k1.type = logger
​
​
#指定sink的类型
a.sinks.k2.type = hdfs
a.sinks.k2.hdfs.path = /data/flume/netcat
# 指定文件名前缀
a.sinks.k2.hdfs.filePrefix = hbaselog
# 指定达到多少数据量写一次文件 单位：bytes
a.sinks.k2.hdfs.rollSize = 102400
# 指定多少条写一次文件
a.sinks.k2.hdfs.rollCount = 1000
# 指定文件类型为 流 来什么输出什么
a.sinks.k2.hdfs.fileType = DataStream
# 指定文件输出格式 为text
a.sinks.k2.hdfs.writeFormat = text
# 指定文件名后缀
a.sinks.k2.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
a.sinks.k2.channel = c1
7、多Agent

vim flume_hbase_regionserver_node1.conf

# 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 -200f /usr/local/soft/hbase-2.2.7/logs/hbase-root-regionserver-node1.log
​
#指定sink的类型
a.sinks.k1.type = avro
a.sinks.k1.hostname = master
a.sinks.k1.port = 7777
​
#指定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

vim flume_hbase_regionserver_node2.conf

# 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 -200f /usr/local/soft/hbase-2.2.7/logs/hbase-root-regionserver-node2.log
​
#指定sink的类型
a.sinks.k1.type = avro
a.sinks.k1.hostname = master
a.sinks.k1.port = 7777
​
#指定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

vim flume_data_to_hdfs_master.conf

# a表示给agent命名为a
# 给source组件命名为r1
a.sources = r1
# 给sink组件命名为k1
a.sinks = k1 
# 给channel组件命名为c1
a.channels = c1
​
#指定exec的属性
a.sources.r1.type = avro 
a.sources.r1.bind = 0.0.0.0 
a.sources.r1.port = 7777
​
#指定sink的类型
a.sinks.k1.type = hdfs
a.sinks.k1.hdfs.path = /data/flume/regionserver/log
# 指定文件名前缀
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 = c1

http://www.kler.cn/news/366389.html

相关文章：

  • 使用 docker 的方式部署 NFS server 提供文件共享能力
  • 分布式理论基础
  • centeros7 编译ffmpeg
  • 什么是域名？什么是泛域名？
  • 【Python爬虫系列】_031.Scrapy_模拟登陆中间件
  • Python爬虫教程：从入门到精通
  • 中国人寿财险青岛市分公司践行绿色金融，助力可持续发展
  • 【mysql】转义字符反斜杠，正则表达式
  • LabVIEW换流变换器智能巡检系统
  • 三周精通FastAPI：13 响应状态码status_code
  • 2024.10月25日- SpringBoot整合Thymeleaf
  • 深度学习杂乱知识
  • 【论文速读】| 攻击图谱：从实践者的角度看生成式人工智能红队测试中的挑战与陷阱
  • Mysql查询表的结构信息 把列名 数据类型 等变成列数据(适用于生成数据库表结构文档) (三)
  • 一分钟学会MATLAB-数值计算
  • 怎样安装 three.js
  • Python依赖库的几种离线安装方法
  • 【Linux】-----进程控制
  • IDEA如何将一个分支的代码合并到另一个分支（当前分支）
  • PyCharm 2023 版本之后使用本地 conda 已存在环境的方法
  • Golang 怎么高效处理ACM模式输入输出
  • 实验一 嵌入式开发基础 4-6学时实践
  • 【ubuntu20联网】ubuntu20.04正常使用网络，解决校园以太网无法连接问题
  • ​​Spring6梳理17——基于XML的自动装配
  • Spring Retry框架
  • 从视频中学习的SeeDo：VLM解释视频并生成规划、代码(含通过RGB视频模仿的人形机器人OKAMI、DexMV)