Flume介绍、安装、使用案例、自定义Source/Sink、监控

电玩女神 2022-10-09 03:13 148阅读 0赞

### 本文目录 ###

*  1.大数据领域的传输问题
 *  2.Flume入门
 *   *  2.1 Flume 基础架构
     *   *  Ⅰ. Agent
         *  Ⅱ.Source
         *  Ⅲ.Sink
         *  Ⅳ.Channel
         *  Ⅴ.Event
 *  3.Flume 安装
 *  4.Flume案例
 *   *  Ⅰ.Flume基础使用
     *   *  1.监控 netcat 端口数据案例
         *  2.实时监控单个追加文件(tail -f)，比如：日志
         *  3.实时监控单个追加文件(tail -f)，Sink至 HDFS，比如：日志
         *  4.实时监控目录下多个新文件
         *  5.实时监控多个目录下的多个追加文件
     *  Ⅱ.Flume进阶使用
     *   *  6.复制通道选择器(单数据源多出口案例，即：1进2出)
         *  7.故障转移
         *  8.负载均衡(与故障转移配置基本相似)
         *  9.聚合(多数据源聚合，即：2进，聚合后，1出)
         *  10.多路复用(需配合自定义Interceptor来使用。即：一个日志文件(多个系统的日志都在该文件中)，根据日志中某个字段值，比如type=1，是系统A日志，sink to hdfs；type=2，是系统B日志，sink to kafka)
 *  5.自定义Source、Sink
 *   *  Ⅰ.自定义Source
     *  Ⅱ.自定义Sink
 *  6.Flume 数据流监控

# 1.大数据领域的传输问题 #

大数据领域，主要解决的就是3件事情：`1.海量数据的存储(hdfs)`  `2.海量数据的传输(??)`   `3.海量数据的计算(MapReduce)`。Flume的出现，主要用来解决`海量数据的传输问题`。

`Java后台服务器日志 到 大数据集群`。生产环境下，Java后台 和 大数据集群是不在同一台服务器。`如何将数据从Java后台传输到大数据集群呢。` `分布式环境下，如何将 N 台服务器的 web 日志上传到大数据集群呢？`

既然不可能将Java web后台和大数据集群放在一台机器上，肯定需要使用工具将数据传输过来。市面上传输工具也有很多，但是Flume用的是最广泛的。生产环境中，数据放在后台服务器的日志中，我们需要将其采集过来，就要用到这个Flume，来做这个数据的采集。

> **场景：**  
>   Java后台 web 日志，每天都在实时产生。一天结束后，日志会以 **xxx\_log\_2021-06-23.out** 这种格式进行保存。此时，我们便可以使用 `hadoop fs -put` 方式,将其上传至 hdfs。这个是没问题的。但是这种情况下，只有等一天结束后日志才能够进行上传。因为put 不能实时一直put。如果要等到第二天，才能把前一天的数据进行上传，显然实时性就无法满足。   
>    
> **问题分析：**  
>   虽然数据是实时产生的，但是只能过了1天后才导到hdfs，比如凌晨1点，然后还要进行分析，加载到hive表，用spark做离线计算等。等数据分析完就到早上了，头一天浏览了什么东西，等到第二天，第三天才去推荐。【现在的推荐系统，都是很有时效性的。日志隔一天上传就不太靠谱了】。  
>    
> **解决：**  
>   我们需要一个中间的组件，帮助我们实时的读取数据，实时的上传的hdfs，实时的处理这些数据。就需要引入接下来介绍的 Flume 了。【Flume 能帮我们解决的就是两个问题：`1.大数据量`   `2.实时传输`】

# 2.Flume入门 #

Flume 是 Cloudera 提供的一个`高可用的`，`高可靠的`，`分布式`的`海量日志`的`采集`、`聚合`和`传输`的系统。Flume 基于流式架构，灵活简单。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6YjM0ODExMDE3NQ_size_16_color_FFFFFF_t_70]  
  根据Flume官网介绍：[http://flume.apache.org][http_flume.apache.org]，Flume就是传输大量日志的，没法传输图片等内容。只能传输日志数据。如下图所示：

![在这里插入图片描述][20210623163656169.png_pic_center]

## 2.1 Flume 基础架构 ##

Flume 组成架构如图所示：  
![在这里插入图片描述][20210623163625374.png_pic_center]  
下面我们来详细介绍一下 Flume 架构中的组件：

### Ⅰ. Agent ###

Agent 是一个 JVM 进程，它以`事件(Event)`的形式将数据从源头送至目的

Agent 主要有 3 个部分组成，`Source`、`Channel`、`Sink`。Agent 就是将来我们要启动的一个JVM进程。

> 这三个组件如何配置，Source 到底读的是文件？还是端口数据？还可以读Kafka数据，有很多源头数据。  
>    
> Sink 可以写到 HDFS、Kafka。到底要写到哪里，`需要我们通过配置来告诉它`。比如说你希望使用 Flume 采集数据到 HDFS 还是到 Kafka，你得告诉它。  
>    
> 所以我们需要配置这三个组件。配置完就是一个配置文件。根据这个配置文件启动一个Agent进程就完成了 Flume 的整个的工作，中途不需要写任何代码，只需要写好配置文件就 ok 了。

### Ⅱ.Source ###

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

### Ⅲ.Sink ###

Sink 不断地轮询 Channel 中的事件且批量地移除它们，并将这些事件批量写入到存储或索引系统、或者被发送到另一个 Flume Agent。

Sink 组件目的地包括 `hdfs`、logger、`avro`、thrift、ipc、`file`、HBase、solr、`自定义Sink`。

### Ⅳ.Channel ###

Channel 是位于 Source 和 Sink 之间的缓冲区。因此，Channel 允许 Source 和 Sink 运作在不同的速率上。Channel 是线程安全的，可以同时处理几个 Source 的写入操作和几个Sink 的读取操作。

Flume 自带几种 Channel：`Memory Channel`、`File Channel` 、`Kafka Channel`、`JDBC Channel` 等。

Memory Channel 是内存中的队列。Memory Channel 在不需要关心数据丢失的情景下适用。如果需要关心数据丢失，那么 Memory Channel 就不应该使用，因为程序死亡、机器宕机或者重启都会导致数据丢失。File Channel 将所有事件写到磁盘。因此在程序关闭或机器宕机的情况下不会丢失数据。

### Ⅴ.Event ###

事件，Flume 数据传输的基本单元，以 Event 的形式将数据从源头送至目的地。`Event 由 Header 和 Body 两部分组成，Header 用来存放该 event 的一些属性，为 K-V 结构，Body 用来存放该条数据，形式为字节数组`。

> **提示：**  
>   其实官方提供的就很多了。自定义Source、Sink 的情况也相对较少。

# 3.Flume 安装 #

> Flume 官网地址：[http://flume.apache.org][http_flume.apache.org 1]  
> Flume 用户文档：[http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html][http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html]  
> Flume 下载地址：[http://archive.apache.org/dist/flume][http_archive.apache.org_dist_flume]

Flume 安装贼简单，只需要解压缩 tar.gz 包，`flume-env.sh` 中配置一个 JAVA\_HOME 即可。

> 1.将 `apache-flume-1.9.0-bin.tar.gz` 上传到 linux的 **/opt/software**目录下  
>    
> 2.解压 `apache-flume-1.9.0-bin.tar.gz` 到 **/opt/module/** 目录下  
>  \[hadoop@hadoop201 software\]$ **tar -zxf apache-flume-1.9.0-bin.tar.gz -C /opt/module/**  
>    
> 3.修改 `apache-flume-1.9.0-bin` 的名称为 **flume**  
>  \[hadoop@hadoop201 module\]$ **mv apache-flume-1.9.0-bin flume**  
>    
> 4.将 `flume/conf` 下的 `flume-env.sh.template` 文件修改为 `flume-env.sh`，并配置 **flume-env.sh** 文件  
>  \[hadoop@hadoop201 conf\]$ **mv flume-env.sh.template flume-env.sh**  
>  \[hadoop@hadoop201 conf\]$ **vi flume-env.sh**  
>  **export JAVA\_HOME=/opt/module/jdk1.8.0\_261**

# 4.Flume案例 #

> **提示：**  
>   编写Flume配置，请参考官网用户手册，地址：[http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html][http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html]

## Ⅰ.Flume基础使用 ##

### 1.监控 netcat 端口数据案例 ###

> **选择：**`netcat source` \+ `memory channel` \+ `logger sink`  
> 地址：[https://blog.csdn.net/lzb348110175/article/details/118179386][https_blog.csdn.net_lzb348110175_article_details_118179386]

### 2.实时监控单个追加文件(tail -f)，比如：日志 ###

> **选择：**`exec source` \+ `memory channel` \+ `logger sink`  
> 地址：[https://blog.csdn.net/lzb348110175/article/details/118182358][https_blog.csdn.net_lzb348110175_article_details_118182358]

### 3.实时监控单个追加文件(tail -f)，Sink至 HDFS，比如：日志 ###

> **选择：**`exec source` \+ `memory channel` \+ `hdfs sink`  
> 地址：[https://blog.csdn.net/lzb348110175/article/details/118186673][https_blog.csdn.net_lzb348110175_article_details_118186673]

### 4.实时监控目录下多个新文件 ###

> **选择：**`spooling directory source` \+ `memory channel` \+ `hdfs sink`  
> 地址：[https://blog.csdn.net/lzb348110175/article/details/118188154][https_blog.csdn.net_lzb348110175_article_details_118188154]

### 5.实时监控多个目录下的多个追加文件 ###

> **选择：**`taildir source` \+ `memory channel` \+ `hdfs sink`  
> 地址：[https://blog.csdn.net/lzb348110175/article/details/118189312][https_blog.csdn.net_lzb348110175_article_details_118189312]

## Ⅱ.Flume进阶使用 ##

### 6.复制通道选择器(单数据源多出口案例，即：1进2出) ###

> **选择：** Flume1：`taildir source` \+ `memory channel` \+ `avro sink + Replicating Channel Selector(复制通道选择器)`  
>     Flume2：`avro source` \+ `memory channel` \+ `hdfs sink`  
>     Flume3：`avro source` \+ `memory channel` \+ `file role sink(写到本地文件系统)`  
> 地址：[https://blog.csdn.net/lzb348110175/article/details/118191189][https_blog.csdn.net_lzb348110175_article_details_118191189]

### 7.故障转移 ###

> **选择：** Flume1：`netcat source` \+ `memory channel` \+ `avro sink + Failover Sink Processor(故障转移接收器处理器)`  
>     Flume2：`avro source` \+ `memory channel` \+ `logger sink`  
>     Flume3：`avro source` \+ `memory channel` \+ `logger sink`  
> 地址：[https://blog.csdn.net/lzb348110175/article/details/118220586][https_blog.csdn.net_lzb348110175_article_details_118220586]

### 8.负载均衡(与故障转移配置基本相似) ###

> **选择：** Flume1：`netcat source` \+ `memory channel` \+ `avro sink + Load Balancing Sink Processor(负载均衡接收器处理器)`  
>     Flume2：`avro source` \+ `memory channel` \+ `logger sink`  
>     Flume3：`avro source` \+ `memory channel` \+ `logger sink`  
> 地址：[https://blog.csdn.net/lzb348110175/article/details/118223517][https_blog.csdn.net_lzb348110175_article_details_118223517]

### 9.聚合(多数据源聚合，即：2进，聚合后，1出) ###

> **选择：** Flume1：`taildir source` \+ `memory channel` \+ `avro sink`  
>     Flume2：`taildir source` \+ `memory channel` \+ `avro sink`  
>     Flume3：`avro source` \+ `memory channel` \+ `hdfs sink`  
> 地址：[https://blog.csdn.net/lzb348110175/article/details/118298263][https_blog.csdn.net_lzb348110175_article_details_118298263]

### 10.多路复用(需配合自定义Interceptor来使用。即：一个日志文件(多个系统的日志都在该文件中)，根据日志中某个字段值，比如type=1，是系统A日志，sink to hdfs；type=2，是系统B日志，sink to kafka) ###

> **选择：** Flume1：`taildir source` \+ `memory channel` \+ `avro sink + Multiplexing Channel Selector(多路复用通道选择器)`  
>     Flume2：`taildir source` \+ `avro channel` \+ `logger sink`  
>     Flume3：`avro source` \+ `avro channel` \+ `logger sink`  
> 地址：[https://blog.csdn.net/lzb348110175/article/details/118306013][https_blog.csdn.net_lzb348110175_article_details_118306013]

# 5.自定义Source、Sink #

## Ⅰ.自定义Source ##

## Ⅱ.自定义Sink ##

# 6.Flume 数据流监控 #

Ganglia

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6YjM0ODExMDE3NQ_size_16_color_FFFFFF_t_70]: /images/20221005/5f813a810191442cbc9abe535e1dce65.png
[http_flume.apache.org]: http://flume.apache.org/
[20210623163656169.png_pic_center]: /images/20221005/2203bac8ed304044aa127f0a99bd23eb.png
[20210623163625374.png_pic_center]: /images/20221005/b0451af941f3459f9c5a7c4539582783.png
[http_flume.apache.org 1]: http://flume.apache.org
[http_flume.apache.org_releases_content_1.9.0_FlumeUserGuide.html]: http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html
[http_archive.apache.org_dist_flume]: http://archive.apache.org/dist/flume
[https_blog.csdn.net_lzb348110175_article_details_118179386]: https://blog.csdn.net/lzb348110175/article/details/118179386
[https_blog.csdn.net_lzb348110175_article_details_118182358]: https://blog.csdn.net/lzb348110175/article/details/118182358
[https_blog.csdn.net_lzb348110175_article_details_118186673]: https://blog.csdn.net/lzb348110175/article/details/118186673
[https_blog.csdn.net_lzb348110175_article_details_118188154]: https://blog.csdn.net/lzb348110175/article/details/118188154
[https_blog.csdn.net_lzb348110175_article_details_118189312]: https://blog.csdn.net/lzb348110175/article/details/118189312
[https_blog.csdn.net_lzb348110175_article_details_118191189]: https://blog.csdn.net/lzb348110175/article/details/118191189
[https_blog.csdn.net_lzb348110175_article_details_118220586]: https://blog.csdn.net/lzb348110175/article/details/118220586
[https_blog.csdn.net_lzb348110175_article_details_118223517]: https://blog.csdn.net/lzb348110175/article/details/118223517
[https_blog.csdn.net_lzb348110175_article_details_118298263]: https://blog.csdn.net/lzb348110175/article/details/118298263
[https_blog.csdn.net_lzb348110175_article_details_118306013]: https://blog.csdn.net/lzb348110175/article/details/118306013