Flume

news2024/11/26 22:17:22

目录

  • 1、Flume简要描述
  • 2、Flume网址
  • 3、Flume的组成架构
    • 3.1、Agent
    • 3.2、Source
    • 3.3、Channel
    • 3.4、Sink
    • 3.5、Event
  • 4、拓扑结构
  • 5、运行原理
  • 6、简要安装
  • 7、开发案例
    • 7.1、监控端口数据官方案例
      • 1)安装telnet工具
      • 2)判断44444端口是否被占用
      • 3)创建Flume Agent配置文件flume-telnet-logger.conf
      • 4)先开启flume监听端口
      • 5)使用telnet工具向本机的44444端口发送内容
      • 6)在Flume监听页面观察接收数据情况
    • 7.2、实时读取本地文件到HDFS案例
    • 7.3、实时读取目录文件到HDFS案例
    • 7.4、单数据源多出口案例
    • 7.5、单数据源多出口案例(Sink组)
    • 7.6、多数据源汇总案例
    • 7.7、Flume高级之自定义MySQLSource
  • 8、Flume监控之Ganglia
    • 8.1、Ganglia的安装与部署
    • 8.2 操作Flume测试监控
  • 9、Flume知识扩展
      • 1、使用第三方框架Ganglia实时监控Flume。
      • 2、Flume参数调优
      • 3、Flume的事务机制
      • 4、Flume的Channel Selectors

1、Flume简要描述

Flume是 Apache提供的一个高可用的,高可靠的,分布式的海量日志采集、聚合和传输的系统。Flume基于流式架构,灵活简单。

2、Flume网址

1、官网地址
2、文档查看地址
3、下载地址
4、github地址

3、Flume的组成架构

在这里插入图片描述
在这里插入图片描述
1、Source数据输入端的常见类型有:avro、spooling、exec、syslog、netcat、kafka等等。

2、put事务流程:

  • doPut:将批数据先写入临时缓冲区putList。
  • doCommit:检查channel内存队列是否足够合并。
  • doRollback:channel内存队列空间不足,回滚数据。

3、channel是位于Source和Sink之间的缓冲区。

Flume自带两种channel:Memory Channel和File Channel。
Memory Channel是基于内存缓存的,在不需要关心数据丢失的情况下适用。
File Channel是Flume的持久化Channel,系统宕机不会丢失数据。
4、Take事务

  • doTake:先将数据取到临时缓冲区takeList
  • doCommit:如果数据全部发送成功,则清楚临时缓冲区takeList
  • doRollBack:数据发送过程中如果出现异常,rollBack将临时缓冲区takeList中的数据归还给Channel内存队列。

5、Sink组件常用的有:HDFS、Kafka、log、avro、File、自定义。

3.1、Agent

Agent是一个JVM进程,它以事件的形式将数据从源头送至目的,是Flume数据传输的基本单元。
Agent主要有3个部分组成,Source、Channel、Sink。

3.2、Source

Source是负责接收数据到Flume Agent的组件。Source组件可以处理各种类型、各种格式的日志数据,包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy。

3.3、Channel

Channel是位于Source和Sink之间的缓冲区。因此,Channel允许Source和Sink运作在不同的速率上。Channel是线程安全的,可以同时处理几个Source的写入操作和几个Sink的读取操作。
Flume自带两种Channel:Memory Channel和File Channel。
Memory Channel是内存中的队列。Memory Channel在不需要关心数据丢失的情景下适用。如果需要关心数据丢失,那么Memory Channel就不应该使用,因为程序死亡、机器宕机或者重启都会导致数据丢失。
File Channel将所有事件写到磁盘。因此在程序关闭或机器宕机的情况下不会丢失数据。

3.4、Sink

Sink不断地轮询Channel中的事件且批量地移除它们,并将这些事件批量写入到存储或索引系统、或者被发送到另一个Flume Agent。
Sink是完全事务性的。在从Channel批量删除数据之前,每个Sink用Channel启动一个事务。批量事件一旦成功写出到存储系统或下一个Flume Agent,Sink就利用Channel提交事务。事务一旦被提交,该Channel从自己的内部缓冲区删除事件。
Sink组件目的地包括hdfs、logger、avro、thrift、ipc、file、null、HBase、solr、自定义。

3.5、Event

传输单元,Flume数据传输的基本单元,以事件的形式将数据从源头送至目的地。

4、拓扑结构

 Flume Agent连接
图1-1 Flume Agent连接
在这里插入图片描述
图1-2 单source,多channel、sink
在这里插入图片描述
图1-3 Flume负载均衡
在这里插入图片描述
图1-4 Flume Agent聚合

5、运行原理

在这里插入图片描述

6、简要安装

1)将apache-flume-1.11.0-bin.tar.gz上传到linux的/opt/software目录下
2)解压apache-flume-1.11.0-bin.tar.gz到/opt/module/目录下

[atguigu@hadoop102 software]$ tar -zxf apache-flume-1.11.0-bin.tar.gz -C /opt/module/

3)修改apache-flume-1.11.0-bin的名称为flume

[atguigu@hadoop102 module]$ mv apache-flume-1.11.0-bin flume

4)将flume/conf下的flume-env.sh.template文件修改为flume-env.sh,并配置flume-env.sh文件

[atguigu@hadoop102 conf]$ mv flume-env.sh.template flume-env.sh
[atguigu@hadoop102 conf]$ vi flume-env.sh
export JAVA_HOME=/opt/module/jdk1.8.0_144

7、开发案例

7.1、监控端口数据官方案例

1)案例需求:首先,Flume监控本机44444端口,然后通过telnet工具向本机44444端口发送消息,最后Flume将监听的数据实时显示在控制台。
在这里插入图片描述

1)安装telnet工具

1、安装telnet
step 1、rpm -qa telnet-server(无输出表示telnet-server未安装,则执行step2;否则执行step3)

step 2、yum -y install telnet-server(安装telnet-server)

step 3、rpm -qa telnet(无输出表示telnet未安装,则执行step4,否则执行step5)
step 4、yum -y install telnet(安装)
二、重新启动xinetd守护进程

由于telnet服务也是由xinetd守护的,所以安装完telnet-server后,需要重新启动xinetd守护进程

step 5、service xinetd restart(如果启动正常直接跳到step7)

step 6、yum -y install xinetd (安装xinetd,然后在继续执行step5)

三、测试

查看tcp的23端口是否正常开启

step 7、netstat -tnl |grep 23 (如果无输出说明未正常启动,则step8;否则整个安装过程完成)

如下图所示则表示启动正常:
step 8、修改 /etc/xinetd.d/telnet 文件,将disable=yes修改为disable=no,并重新执行step5

如果/etc/ xinetd.d/ 该路径下没有telnet文件,则构造一个telnet文件。
telnet文件内容如下:

# default: on
# description: The telnet server serves telnet sessions; it uses \
#       unencrypted username/password pairs for authentication.
service telnet
{
        disable = yes
        flags           = REUSE
        socket_type     = stream
        wait            = no
        user            = root
        server          = /usr/sbin/in.telnetd
        log_on_failure  += USERID

}

2)判断44444端口是否被占用

[pgxl@hadoop102 flume-telnet]$ sudo netstat -tunlp | grep 44444

功能描述:netstat命令是一个监控TCP/IP网络的非常有用的工具,它可以显示路由表、实际的网络连接以及每一个网络接口设备的状态信息。
基本语法:netstat [选项]
选项参数:
-t或–tcp:显示TCP传输协议的连线状况;
-u或–udp:显示UDP传输协议的连线状况;
-n或–numeric:直接使用ip地址,而不通过域名服务器;
-l或–listening:显示监控中的服务器的Socket;
-p或–programs:显示正在使用Socket的程序识别码和程序名称;

3)创建Flume Agent配置文件flume-telnet-logger.conf

在flume目录下创建job文件夹并进入job文件夹。

[pgxl@hadoop102 flume]$ mkdir job
[pgxl@hadoop102 flume]$ cd job/

在job文件夹下创建Flume Agent配置文件flume-telnet-logger.conf。

[pgxl@hadoop102 job]$ touch flume-telnet-logger.conf

在flume-telnet-logger.conf文件中添加如下内容。

[pgxl@hadoop102 job]$ vim flume-telnet-logger.conf

添加内容如下:

#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

注:配置文件来源于官方手册http://flume.apache.org/FlumeUserGuide.html
在这里插入图片描述

4)先开启flume监听端口

[pgxl@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume-telnet-logger.conf -Dflume.root.logger=INFO,console

参数说明:
–conf conf/ :表示配置文件存储在conf/目录
–name a1 :表示给agent起名为a1
–conf-file job/flume-telnet.conf :flume本次启动读取的配置文件是在job文件夹下的flume-telnet.conf文件。
-Dflume.root.logger==INFO,console :-D表示flume运行时动态修改flume.root.logger参数属性值,并将控制台日志打印级别设置为INFO级别。日志级别包括:log、info、warn、error。

5)使用telnet工具向本机的44444端口发送内容

[pgxl@hadoop102 ~]$ telnet localhost 44444

6)在Flume监听页面观察接收数据情况

在这里插入图片描述

7.2、实时读取本地文件到HDFS案例

7.3、实时读取目录文件到HDFS案例

7.4、单数据源多出口案例

7.5、单数据源多出口案例(Sink组)

7.6、多数据源汇总案例

7.7、Flume高级之自定义MySQLSource

8、Flume监控之Ganglia

8.1、Ganglia的安装与部署

  1. 安装httpd服务与php

[pgxl@hadoop102 flume]$ sudo yum -y install httpd php

  1. 安装其他依赖

[pgxl@hadoop102 flume]$ sudo yum -y install rrdtool perl-rrdtool rrdtool-devel
[pgxl@hadoop102 flume]$ sudo yum -y install apr-devel

  1. 安装ganglia

[pgxl@hadoop102 flume]$ sudo rpm -Uvh http://dl.fedoraproject.org/pub/epel/6/x86_64/epel-release-6-8.noarch.rpm
[pgxl@hadoop102 flume]$ sudo yum -y install ganglia-gmetad
[pgxl@hadoop102 flume]$ sudo yum -y install ganglia-web
[pgxl@hadoop102 flume]$ sudo yum install -y ganglia-gmond

  1. 修改配置文件ganglia.conf

[pgxl@hadoop102 flume]$ sudo vim /etc/httpd/conf.d/ganglia.conf

修改为红颜色的配置:

#Ganglia monitoring system php web frontend
Alias /ganglia /usr/share/ganglia
<Location /ganglia>
Order deny,allow
Deny from all
Allow from all
# Allow from 127.0.0.1
# Allow from ::1
#Allow from .example.com

  1. 修改配置文件gmetad.conf

[pgxl@hadoop102 flume]$ sudo vim /etc/ganglia/gmetad.conf

修改为:

data_source “hadoop102” 192.168.1.102

  1. 修改配置文件gmond.conf

[pgxl@hadoop102 flume]$ sudo vim /etc/ganglia/gmond.conf

修改为:

cluster {
name = “hadoop102”
owner = “unspecified”
latlong = “unspecified”
url = “unspecified”
}
udp_send_channel {
#bind_hostname = yes # Highly recommended, soon to be default.
# This option tells gmond to use a source address
# that resolves to the machine’s hostname. Without
# this, the metrics may appear to come from any
# interface and the DNS names associated with
# those IPs will be used to create the RRDs.
#mcast_join = 239.2.11.71
host = 192.168.1.102
port = 8649
ttl = 1
}
udp_recv_channel {
#mcast_join = 239.2.11.71
port = 8649
bind = 192.168.1.102
retry_bind = true
#Size of the UDP buffer. If you are handling lots of metrics you really
#should bump it up to e.g. 10MB or even higher.
#buffer = 10485760
}

  1. 修改配置文件config

[pgxl@hadoop102 flume]$ sudo vim /etc/selinux/config
修改为:
#This file controls the state of SELinux on the system.
#SELINUX= can take one of these three values:
#enforcing - SELinux security policy is enforced.
#permissive - SELinux prints warnings instead of enforcing.
#disabled - No SELinux policy is loaded.
SELINUX=disabled
#SELINUXTYPE= can take one of these two values:
#targeted - Targeted processes are protected,
#mls - Multi Level Security protection.
SELINUXTYPE=targeted

提示:selinux本次生效关闭必须重启,如果此时不想重启,可以临时生效之:

[pgxl@hadoop102 flume]$ sudo setenforce 0

  1. 启动ganglia

[pgxl@hadoop102 flume]$ sudo service httpd start
[pgxl@hadoop102 flume]$ sudo service gmetad start
[pgxl@hadoop102 flume]$ sudo service gmond start

  1. 打开网页浏览ganglia页面
    http://192.168.1.102/ganglia
    提示:如果完成以上操作依然出现权限不足错误,请修改/var/lib/ganglia目录的权限:

[pgxl@hadoop102 flume]$ sudo chmod -R 777 /var/lib/ganglia

8.2 操作Flume测试监控

  1. 修改/opt/module/flume/conf目录下的flume-env.sh配置:

JAVA_OPTS=“-Dflume.monitoring.type=ganglia
-Dflume.monitoring.hosts=192.168.1.102:8649
-Xms100m
-Xmx200m”

  1. 启动Flume任务

[pgxl@hadoop102 flume]$ bin/flume-ng agent
–conf conf/
–name a1
–conf-file job/flume-telnet-logger.conf
-Dflume.root.logger==INFO,console
-Dflume.monitoring.type=ganglia
-Dflume.monitoring.hosts=192.168.1.102:8649

  1. 发送数据观察ganglia监测图

[pgxl@hadoop102 flume]$ telnet localhost 44444

样式如图:
在这里插入图片描述
图例说明:
在这里插入图片描述

9、Flume知识扩展

1、使用第三方框架Ganglia实时监控Flume。

2、Flume参数调优

  1. Source
    增加Source个(使用Tair Dir Source时可增加FileGroups个数)可以增大Source的读取数据的能力。例如:当某一个目录产生的文件过多时需要将这个文件目录拆分成多个文件目录,同时配置好多个Source 以保证Source有足够的能力获取到新产生的数据。
    batchSize参数决定Source一次批量运输到Channel的event条数,适当调大这个参数可以提高Source搬运Event到Channel时的性能。
  2. Channel
    type 选择memory时Channel的性能最好,但是如果Flume进程意外挂掉可能会丢失数据。type选择file时Channel的容错性更好,但是性能上会比memory channel差。
    使用file Channel时dataDirs配置多个不同盘下的目录可以提高性能。
    Capacity 参数决定Channel可容纳最大的event条数。transactionCapacity 参数决定每次Source往channel里面写的最大event条数和每次Sink从channel里面读的最大event条数。transactionCapacity需要大于Source和Sink的batchSize参数。
  3. Sink
    增加Sink的个数可以增加Sink消费event的能力。Sink也不是越多越好够用就行,过多的Sink会占用系统资源,造成系统资源不必要的浪费。
    batchSize参数决定Sink一次批量从Channel读取的event条数,适当调大这个参数可以提高Sink从Channel搬出event的性能。

3、Flume的事务机制

Flume的事务机制(类似数据库的事务机制):Flume使用两个独立的事务分别负责从Soucrce到Channel,以及从Channel到Sink的事件传递。比如spooling directory source 为文件的每一行创建一个事件,一旦事务中所有的事件全部传递到Channel且提交成功,那么Soucrce就将该文件标记为完成。同理,事务以类似的方式处理从Channel到Sink的传递过程,如果因为某种原因使得事件无法记录,那么事务将会回滚。且所有的事件都会保持到Channel中,等待重新传递。

4、Flume的Channel Selectors

在这里插入图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/88789.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

线索二叉树构建和遍历

线索二叉树 文章目录线索二叉树引出线索解决方案(以中序线索树为例)引论:策略&#xff1a;问题1&#xff1a;♥ 对策线索二叉树的节点类型定义线索化二叉树♥ 线索化算法♥ 建立中序线索二叉树的算法♥ 算法实现遍历线索化二叉树引言:♥ 遍历思路(中序遍历举例)&#xff1a;♥ …

产品设计:Material Design 学习笔记一

​自从Material Design发布以来&#xff0c;我就在一直收集相关素材与资源&#xff0c;研究别人的作品。这套设计风格非常鲜明&#xff0c;带有浓郁的Google式严谨和理性哲学&#xff0c;深得我心。实际上&#xff0c;光是研究素材和别人作品&#xff0c;就能发现一些明显的规律…

【Docker】虚悬镜像(Dangling Image)介绍和处理方法

本期目录1. 虚悬镜像介绍2. 查看本地所有虚悬镜像3. 删除全部虚悬镜像4. 人为构建虚悬镜像专栏精选文章1. 虚悬镜像介绍 虚悬镜像 (Dangling Image) 指的是仓库名 (镜像名) 和标签 TAG 都是 <none> 的镜像。如下图所示。 在实际开发中&#xff0c;在构建或者删除镜像时出…

【案例实践】气象数据相关分析及使用系列:如何使用格点数据分析中国积温变化技术应用

【视频教程】气象数据相关分析及使用系列课程&#xff1a;如何使用格点数据分析中国积温变化技术应用https://mp.weixin.qq.com/s?__bizMzAxNzcxMzc5MQ&mid2247519871&idx8&sn5ccca4436825438ce253ab7455437259&chksm9be3916fac94187970353477a4c191cdb6d499d…

[LINUX]基本权限

&#x1f941;作者&#xff1a; 华丞臧. &#x1f4d5;​​​​专栏&#xff1a;【LINUX】 各位读者老爷如果觉得博主写的不错&#xff0c;请诸位多多支持(点赞收藏关注)。如果有错误的地方&#xff0c;欢迎在评论区指出。 推荐一款刷题网站 &#x1f449; LeetCode刷题网站 文…

web前端网页设计与制作——华夏第一县HTML+CSS+JavaScript

家乡旅游景点网页作业制作 网页代码运用了DIV盒子的使用方法&#xff0c;如盒子的嵌套、浮动、margin、border、background等属性的使用&#xff0c;外部大盒子设定居中&#xff0c;内部左中右布局&#xff0c;下方横向浮动排列&#xff0c;大学学习的前端知识点和布局方式都有…

【C++】C++入门

屏幕前的你&#xff0c;一起加油啊&#xff01;&#xff01;&#xff01; 文章目录一、命名空间&#xff08;namespace&#xff09;1.命名空间的定义&#xff08;::域作用限定符&#xff09;2.命名空间的使用&#xff08;三种使用方式&#xff09;二、C输入&输出&#xff…

Maven pom.xm javafx-maven-plugin打包javafx 应用及调试

1、添加 javafx-maven-plugin到 plugin内 <plugin><groupId>com.zenjava</groupId><artifactId>javafx-maven-plugin</artifactId><version>8.8.3</version><configuration><!-- 启动类 --><mainClass>com.test.d…

【春招必备】Java面试题,面试加分项,从jvm层面了解线程的启动和停止

前言 Spring 作为一个轻量级的 Java 开发框架&#xff0c;将面向接口的编程思想贯穿整个 Java 系统应用&#xff0c;因此在 Java 面试中常被提。本次介绍的主要是解析面试过程中如果从源码角度分析常见的问题&#xff0c;为了方便大家阅读&#xff0c;小编这里还整理了一份微服…

零拷贝技术面试题

文章目录1 零拷贝的介绍2 传统的IO的执行流程3 零拷贝涉及的技术点3.1 内核空间和用户空间3.2 用户态和内核态3.3 DMA技术4 零拷贝实现的几种方式4.1 mmapwrite4.2 sendfile4.3 sendfileDMA scatter/gather实现的零拷贝5 java提供的零拷贝方式5.1 Java NIO对mmap的支持5.2 Java…

抖音快手如何轻松接入虚拟人直播

在上一篇文章零基础开启元宇宙——创建虚拟形象中&#xff0c;我们实现了创建虚拟形象&#xff0c;接下来我们可以利用虚拟形象“为所欲为”。今天我们利用虚拟形象在短视频平台如快手、抖音中直播&#xff0c;对于不希望露脸的主播们这是可是一大利器呀&#xff01;话不多说&a…

【Unity项目实战】手把手教学:飞翔的小鸟(5)背景滚动

承接上一篇&#xff1a;【Unity项目实战】手把手教学&#xff1a;飞翔的小鸟&#xff08;4&#xff09;文本添加&#xff0c;我们已经使得主角小鸟接触到地面后跳转到Game Over状态&#xff0c;接下来我们将继续往下&#xff0c;讲解得分机制。 一、重新进入游戏 根据上篇最后…

网络流量回溯分析助力企业实现高效率运维(一)

背景 汽车配件电子图册系统是某汽车集团的重要业务系统。业务部门反映&#xff0c;汽车配件电子图册调用图纸时&#xff0c;出现访问慢现象。 某汽车总部已部署NetInside流量分析系统&#xff0c;使用流量分析系统提供实时和历史原始流量。本次分析重点针对汽车配件电子图册系…

[附源码]Node.js计算机毕业设计二手书交易软件设计与实现Express

项目运行 环境配置&#xff1a; Node.js最新版 Vscode Mysql5.7 HBuilderXNavicat11Vue。 项目技术&#xff1a; Express框架 Node.js Vue 等等组成&#xff0c;B/S模式 Vscode管理前后端分离等等。 环境需要 1.运行环境&#xff1a;最好是Nodejs最新版&#xff0c;我…

数据技术之数据挖掘

第7章 数据挖掘 1.什么是数据挖掘 数据挖掘(Data Mining)就是从大量的数据中&#xff0c;提取隐藏在其中的&#xff0c;事先不知道的、但潜在有用的信息的过程。数据挖掘的目标是建立一个决策模型&#xff0c;根据过去的行动数据来预测未来的行为。 2.阿里数据挖掘平台 阿里…

Java-1213

JVM历程 Sun Classic VM 1996年发布&#xff0c;世界上第一款商用Java虚拟机&#xff0c;JDK1.4时被淘汰&#xff0c;现在hotspot内置了此虚拟机 这款虚拟机只提供了解释器&#xff08;现在主流的虚拟机还会提供即时编译器JIT&#xff09;解释器和JIT两者用一个就可以让程序执…

分享一种 ConstraintLayout 让TextView 自适应的同时,还不超出限制范围的方式

分享一种 ConstraintLayout 让TextView 自适应的同时&#xff0c;还不超出限制范围的方式 不知道大家有没有遇到这种布局需求&#xff1a; 上图布局很简单&#xff0c;ImageView 中间的TextView View ImageView&#xff0c;需求是中间的 TextView 宽度需要根据内容来展示&…

OpenMLDB 实时引擎性能测试报告

OpenMLDB 提供了一个线上线下一致性的特征平台。其中&#xff0c;为了支持低延迟高并发的在线实时特征计算&#xff0c;OpenMLDB 设计实现了一个高性能的实时 SQL 引擎。本报告覆盖了 OpenMLDB 实时 SQL 引擎的性能测试&#xff0c;包含了在较为复杂的负载、典型配置下的各种性…

多线程知识笔记(四)-----volatile、wait方法、notify方法

文章目录1、volatile关键字2、volatile和synchronized对比3、wait和notify方法1、volatile关键字 先看例子&#xff1a; class Counter {public int flag 0; }public class Test4{public static void main(String[] args) {Counter counter new Counter();Thread t1 new Th…

如何使用Footrace 钱包监控功能和设置自定义的交易警报

2022-06-12 本文将介绍如何使用 Footrace 监控 CEX 的钱包地址并设置自定义警报。 什么是 Footrace&#xff1f; Footrace (Foot Trace) 是一个多链的钱包追踪监控平台&#xff0c;可以监控CEX、DEX、鲸鱼、聪明钱、或任何你想关注的地址的钱包。 Footrace 帮助投资者保护他…