【Flume实操】多路复用：实时监听 HTTP 端口数据到控制台案例分析

最新推荐文章于 2024-12-06 16:09:40 发布
最新推荐文章于 2024-12-06 16:09:40 发布
阅读量855 收藏 28
点赞数 18
分类专栏： 人工智能 文章标签： flume http 大数据
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://blog.csdn.net/qq_23934063/article/details/140882621
版权

多路复用：实时监听 HTTP 端口数据到控制台案例分析
案例需求：假设有一个生产场景，Flume1 在持续监控一个 HTTP 端口的数据流，要求根据 Event 中 Header 的某个 key 的值，将不同的 Event 发送到不同的 Channel 中，然后 Flume2 和 Flume3 将获取的数据通过 Sink 端分别写出到控制台。

需求分析：在实际开发中，一台服务器产生的日志类型可能有很多种，不同类型的日志可能需要发送到不同的 Channel。此时会用到 Flume 拓扑结构中的 Multiplexing（多路复用）结构，Multiplexing 结构的原理是， 根据 Event 中 Header 的某个 key 的值，将不同的 Event 发送到不同的 Channel 中，所以我们需要使用 Multiplexing Channel Selector（多路复用通道选择器）。

在该实训中，Flume1 监听本机 4141 端口的数据流，我们以端口数据模拟日志，如果 Event headers 中 key 的值为“Flume”就发往 mc1 通道，如果 Event headers 中 key 的值为“Hadoop”就发往 mc2 通道，然后 Flume2 和 Flume3 分别获取 mc1 和 mc2 的数据并分别写出到控制台。过程如下图所示：
在这里插入图片描述

要求：配置 HTTP Source 监控端口 4141 数据流，配置 Avro Sink 输出数据到下一级 Flume。

flume1

  1. 创建 Agent 配置文件（HTTP Source + File Channel + Avro Sink）
    进入 /root/software/apache-flume-1.9.0-bin/conf 目录，使用 touch 命令创建一个名为 flume1_http_avro.conf 的配置文件。

  2. 查看端口
    使用 netstat -nlp | grep 4141 命令查看 4141、6666 和 6667 端口是否被占用。

  3. 编辑配置文件 flume1_http_avro.conf
    使用 vim 命令打开 flume1_http_avro.conf 文件，在里面添加如下配置：

（1）配置 Flume Agent——a1

首先，我们需要为 a1 命名/列出组件，将数据源（Source）、缓冲通道（Channel）和接收器（Sink）分别命名为 httpSource、 mc1 、 mc2 、 avroSink1 和 avroSink2。

（2）描述和配置 HTTP Source

HTTP Source 可以通过 HTTP POST 和 GET 请求方式接收 Events 数据，GET 通常只能用于测试使用。HTTP 请求会被实现了 HTTPSourceHandler 接口的 handler（处理器）可插拔插件转成 Flume Events，这个 handler 接收一个 HttpServletRequest 后并返回一个 Flume Events 列表。从一次 HTTP 请求解析出来的若干个 Events 会以一个事务提交到 Channel，从而在诸如 File Channel 的一些 Channel 上提高效率。如果 handler 抛出异常，这个 HTTP 的响应状态码是 400。如果 Channel 满了或者无法发送 Events 到 Channel，此时会返回 HTTP 状态码503（服务暂时不可用）。

我们需要为 httpSource 设置以下属性：

属性名称 描述 设置值
channels 与 Source 绑定的 Channel mc1 mc2
type 数据源的类型 http
port 要监听的服务端口 4141（任意可用端口）
bind 要监听的主机名或 IP 地址 localhost（本机）
selector.type 配置 Channel 选择器。可选值：replicating 或 multiplexing，分别表示： 复制、多路复用 multiplexing
selector.header 配置 Event headers 中的 key subject
selector.mapping.Flume 配置 Event headers 中 key 对应的 value，如果是“Flume”就发往 mc1 mc1
selector.mapping.Hadoop 配置 Event headers 中 key 对应的 value，如果是“Hadoop”就发往 mc2 mc2
（3）描述和配置 Avro Sink

Avro Sink 可以作为 Flume 分层收集特性的下半部分。发送到此 Sink 的 Flume Event 将转换为 Avro Event 发送到配置的主机/端口上。Event 从配置的 Channel 中批量获取，数量根据配置的 batch-size 而定。

我们需要为 avroSink1 设置以下属性：

属性名称 描述 设置值
channel 与 Sink 绑定的 Channel mc1
type 接收器的类型 avro
hostname 要监听的主机名或 IP 地址 localhost（本机）
port 要监听的服务端口 6666（任意可用端口）
我们需要为 avroSink2 设置以下属性：

属性名称 描述 设置值
channel 与 Sink 绑定的 Channel mc2
type 接收器的类型 avro
hostname 要监听的主机名或 IP 地址 localhost（本机）
port 要监听的服务端口 6667（任意可用端口）
（4）描述和配置 File Channel

File Channel 是 Flume 的持久通道，它将所有的 Event 写入磁盘，因此不会丢失进程或机器关机、崩溃时的数据，可以很好的保证数据的完整性和一致性。File Channel 通过在一次事务中提交多个 Event 来提高吞吐量，做到了只要事务被提交，那么数据就不会有丢失。在具体配置 File Channel 时，建议 File Channel 设置的目录和程序日志文件保存的目录设成不同的磁盘，以便提高效率。

我们需要为 mc1 设置以下属性：

属性名称 描述 设置值
type 缓冲通道的类型 file
checkpointDir 存储检查点文件的目录，默认值：~/.flume/file-channel/checkpoint /root/flumedata/mc1_checkpoint
dataDirs 用于存储日志文件的目录，可以使用多个，多目录之间使用逗号分隔。在不同物理磁盘上使用多个目录可以提高文件通道的性能。默认值：~/.flume/file-channel/data /root/flumedata/mc1_data
我们需要为 mc2 设置以下属性：

属性名称 描述 设置值
type 缓冲通道的类型 file
checkpointDir 存储检查点文件的目录，默认值：~/.flume/file-channel/checkpoint /root/flumedata/mc2_checkpoint
dataDirs 用于存储日志文件的目录，可以使用多个，多目录之间使用逗号分隔。在不同物理磁盘上使用多个目录可以提高文件通道的性能。默认值：~/.flume/file-channel/data /root/flumedata/mc2_data

a1.sources = httpSource
a1.channels = mc1 mc2
a1.sinks = avroSink1 avroSink2

a1.sources.httpSource.channels = mc1 mc2
a1.sources.httpSource.type = http
a1.sources.httpSource.bind = localhost
a1.sources.httpSource.port = 4141
a1.sources.httpSource.selector.type = multiplexing
a1.sources.httpSource.selector.header = subject
a1.sources.httpSource.selector.mapping.Flume = mc1
a1.sources.httpSource.selector.mapping.Hadoop = mc2

a1.sinks.avroSink1.channel = mc1
a1.sinks.avroSink1.type = avro
a1.sinks.avroSink1.hostname = localhost
a1.sinks.avroSink1.port = 6666

a1.sinks.avroSink2.channel = mc2
a1.sinks.avroSink2.type = avro
a1.sinks.avroSink2.hostname = localhost
a1.sinks.avroSink2.port = 6667

a1.channels.mc1.type = file
a1.channels.mc1.checkpointDir = /root/flumedata/mc1_checkpoint
a1.channels.mc1.dataDirs = /root/flumedata/mc1_data

a1.channels.mc2.type = file
a1.channels.mc2.checkpointDir = /root/flumedata/mc2_checkpoint
a1.channels.mc2.dataDirs = /root/flumedata/mc2_data

vim flume2_avro_logger.conf

a2.sources = avroSource
a2.channels = memoryChannel
a2.sinks = loggerSink

a2.sources.avroSource.channels = memoryChannel
a2.sources.avroSource.type = avro
a2.sources.avroSource.bind = localhost
a2.sources.avroSource.port = 6666

a2.sinks.loggerSink.channel = memoryChannel
a2.sinks.loggerSink.type = logger

a2.channels.memoryChannel.type = memory
a2.channels.memoryChannel.capacity = 1000
a2.channels.memoryChannel.transactionCapacity = 100

vim flume3_avro_logger.conf

a3.sources = avroSource
a3.channels = memoryChannel
a3.sinks = loggerSink

a3.sources.avroSource.channels = memoryChannel
a3.sources.avroSource.type = avro
a3.sources.avroSource.bind = localhost
a3.sources.avroSource.port = 6667

a3.sinks.loggerSink.channel = memoryChannel
a3.sinks.loggerSink.type = logger

a3.channels.memoryChannel.type = memory
a3.channels.memoryChannel.capacity = 1000
a3.channels.memoryChannel.transactionCapacity = 100

flume-ng agent -c conf/ -f conf/flume3_avro_logger.conf -n a3 -Dflume.root.logger=INFO,console
flume-ng agent -c conf/ -f conf/flume2_avro_logger.conf -n a2 -Dflume.root.logger=INFO,console
flume-ng agent -c conf/ -f conf/flume1_http_avro.conf -n a1 -Dflume.root.logger=INFO,console
  1. 往监听端口发送数据
    打开另一个终端，使用 curl -X POST -d 命令发起 POST 请求并往监听端口发送数据。具体命令如下所示：
curl -X POST -d '[{ "headers" :{"subject" : "Flume"},"body" : "shiny"}]' http://localhost:4141
curl -X POST -d '[{ "headers" :{"subject" : "Flume"},"body" : "angel"}]' http://localhost:4141
curl -X POST -d '[{ "headers" :{"subject" : "Hadoop"},"body" : "mark"}]' http://localhost:4141
curl -X POST -d '[{ "headers" :{"subject" : "Hadoop"},"body" : "cendy"}]' http://localhost:4141
curl -X POST -d '[{ "headers" :{"subject" : "BigData"},"body" : "judy"}]' http://localhost:4141
  1. 查看采集结果
    （1）Flume Agent a1 终端
    在启动源发送的代理终端查看控制台输出。如下图所示：
    image.png

从上图可以看出，Flume 已经准确监听并采集到了监听应用发送的 HTTP 请求数据，并成功将接收的所有 Events 写入磁盘。

（2）Flume Agent a2 终端
在启动源第一个接事件的代理终端查看控制台输出。如下图所示：
image.png

这里可以清楚的看到，这个接事件代理只收到了2个事件，因为我们设置了多路复用，只接收 Event headers 中 value 为 “Flume”的事件。

（3）Flume Agent a3 终端
在启动源第二个接事件的代理终端查看控制台输出。如下图所示：
image.png

从上图可以看出，这个接事件代理接收了 Event headers 中 value 为 “Hadoop”的事件。另外，Event headers 中 value 为 “BigData”的事件2个接事件代理都没有接收。

【大数据之路7】数据采集工具 Flume 的使用
程序员五哥
06-14 1851
Apache Flume 是一个分布式的、可靠、高可用的海量日志采集、聚合和传输的系统，支持在系统中定制各类的数据发送方，用于采集数据，同时 Flume 提供对数据的简单处理，并写到各种数据接收方的能力。Flume 和 Sqoop 同属于数据采集系统组件，但Sqoop ⽤来采集关系型数据库数据，⽽ Flume ⽤来采集流动型数据。Flume 名字来源于原始的近乎实战的⽇志数据采集⼯具，现被⼴泛⽤于任何流事件数据的采集，它⽀持从很多数据源聚合数据到 HDFS。
flume详解 从原理到应用
weixin_41929239的博客
11-17 1693
第一章 flume概述 1.1 flume定义 flume是cloudera提供的一个高可用，高可靠，分布式的海量日志采集、聚合和传输的系统。flume基于流式架构，灵活简单。 flume最主要的作用：实时读取服务器本地磁盘的数据，将数据写入HDFS或kafka消息队列中。 1.2 flume基础架构 flume的组成架构如图所示： 1.2.1 agent agent是一个jvm进程，他以事件的形式将数据从源头送至目的地。 agent主要有3个部分组成：source、channel、sink 1.2
flume--多路复用
suojie的博客
01-21 1964
同一时刻，source只能传输给一个channel source是通过event header来决定传输到哪一个channel source 本地 channel c1 localfile c2 mem sink k1 logger k2本地目录 如果文件名1_txt 传输到logger 如果是2_txt 传输到本地 a1.sources = s1 a1.channels =...
Flume学习(三)Flume的多路复用
秘密的博客
12-21 5030
目录多路复用概述准备工作标题 多路复用概述 准备工作 标题
flume多路复用/故障转移/负载均衡/日志聚合
cevent的博客
06-13 1678
一、多路复用 （1）conf配置 [cevent@hadoop207 ~]$ cd /opt/module/apache-flume-1.7.0/ [cevent@hadoop207 apache-flume-1.7.0]$ ll 总用量 176 drwxr-xr-x. 2 cevent cevent 4096 6月 11 13:35 bin -rw-r--r--. 1 cevent cevent 77387 10月 11 2016 CHANGELOG drw
Flume的多路复用案例 *
张俊杰 的博客
10-24 552
需求 Flume1监控SpringBoot的日志信息, 然后将监控到的内容传递给Flume2和Flume3 , Flume2 和Flume3 可以各自处理各自的逻辑,比如说Flume2 上传到HDFS上, Flume3 可以上传到Kafka里面… 这里演示Flume2和Flume3就直接通过打印日志的形式演示了. 找一个SpringBoot项目 随便找一个SpringBoot项目, 注意,要在启动zjj101的agent之前启动这个SpringBoot项目 可以用我这个: https://blog.csd
Flume案例十：多路复用通道选择器(Multiplexing Channel Selector)
扛麻袋的少年的博客
06-28 2427
Flume案例十：多路复用通道选择器(Multiplexing Channel Selector) 选型：  Flume-1：taildir source + memory channel + avro sink + Multiplexing Channel Selector(多路复用渠道选择器)  Flume-2：avro source + memory channel + logger sink  Flume-3：avro source + memory channel + logger sink
大数据求索(6): 使用Flume进行数据采集单机监控端口、监控文件、跨节点等多种方式
腾云
12-09 1106
大数据求索(6): 使用Flume进行数据采集 大数据最好的学习资料是官方文档。 Flume官方文档地址http://flume.apache.org/ Flume简单介绍 Flume是一种分布式的、可靠的且可用的服务，用于高效收集、聚合和移动大量日志数据。它具有基于流式数据的简单灵活架构。它具有可靠性机制和许多故障转移和恢复机制，具有强大的容错能力。它使用简单的可扩展数据模型，允许在线分析应...
Flume系列：案例-Flume 聚合拓扑(常见的日志收集结构)
wester的博客
05-28 1035
系统环境：centos7Java环境：Java8案例只演示通道流程，其中Source，channel，Sink的种类按需调整。
Flume简介及基本运用（含案例）
小财迷嘻嘻的博客
12-10 723
目录1、背景2、Flume简介3、Flume架构和基本概念3.1 基本架构3.2 基本概念3.3 组件种类4、Flume架构模式4.1 单一流程4.2 多代理流程（多个agent顺序连接）4.3 流的合并（多个Agent的数据汇聚到同一个Agent ）4.4 多路复用流（多级流）4.5 load balance功能5、Flume配置格式6、Flume使用案例6.1 案例一 1、背景 2、Flume简介 Flume是日志收集框架，详细来说：Flume是一个分布式、可靠、高可用的海量日志聚合系统，支持在系统中
flume多路复用及拦截器的使用
qq_43224174的博客
01-07 1137
需求 使用flume采集服务器本地日志，需要按照日志类型的不同来将不同种类的日志发往不同的分析系统 需求分析及原理 在开发当中，一台服务器产生的日志类型往往是有很多种的，在不同类型的日志有可能要发往不同的分析系统，在这个时候就会用到flume的channel selector中的Multiplexing结构。其原理是：根据event中header的某个key值，将不同的event发送到不同的channel中，所以我们需要自定义一个interceptor，来为不同类型的event的header中的key赋不同
二百一十五、Flume——Flume拓扑结构之复制和多路复用的开发案例（亲测，附截图）
tiantang2renjian的博客
12-12 1442
Flume拓扑结构之复制和多路复用的开发案例（亲测，附截图）
flume多路复用模式
qq_38130667的博客
11-06 723
Flume经典案例之复制和多路复用
wanjialin的博客
12-26 557
准备工作： 1、准备一个文件，方便后面使用 2、Hadoop集群，需要三个节点 操作步骤： 1、创建配置文件agent1.conf，分别在node1，node2，node3中 内容如下： #agent1(node1) a1.sources=r1 a1.channels=c1 c2 a1.sinks=k1 k2 a1.sources.r1.type=exec a1.sources.r1.command=tail -F /opt/soft/flume190/demo/123.log #配置channelSel
Flume——进阶（agent特性+三种结构：串联，多路复用，聚合）
最新发布
wusuoweiieq的博客
12-06 1023
这种模式是将多个flume顺序连接起来了，从最初的source开始到最终sink传送的目的存储系统。此模式不建议桥接过多的flume数量， flume数量过多不仅会影响传输速率，而且一旦传输过程中某个节点flume宕机，会影响整个传输系统。Flume支持将事件流向一个或者多个目的地。这种模式可以将相同数据复制到多个channel中，或者将不同数据分发到不同的channel中，sink可以选择传送到不同的目的地。详细可以参考上面的Agent ChannelSelector和SinkProcessor。
Flume快速入门系列(4) | 多路复用
不温卜火
05-12 8682
  上一篇我们已经介绍了如何实时读取本地/目录文件到HDFS上，那么这一篇文章博主继续为大家介绍如何实时读取本地/目录文件到HDFS上。 目录一. 单数据源多出口案例1.1 选择器1. 案例需求2. 需求分析3. 实现步骤1. 准备工作2. 创建flume-file-flume.conf3. 创建flume-flume-hdfs.conf4. 创建flume-flume-dir.conf5. 执行...
Flume复制和多路复用原理与实现
FlatTiger的博客
02-24 1149
原理 复制：将相同的数据发送到不同Channel中，不同Channel中的数据分发到不同的Sink，不同的Sink再将数据发送到指定位置。 多路复用：利用拦截器将不同的数据发送到不同的Channel中，不同Channel中的数据分发到不同的Sink，不同的Sink再将数据发送到指定位置。 案例 复制实现 # agent1配置文件放在hadoop01 # 定义sources、channels、sinks a1.sources = r1 a1.channels = c1 c2 a1.sinks = k1 k
flume复制和多路复用
weixin_42908473的博客
10-30 224
命令容易忘 bin/flume-ng agent -n a1 -c conf/ -f job/f1.conf -Dflume.root.logger=INFO,console flume1配置： 1.定义agent，source，channel，sink的名称 a1.sources=r1 a1.channels=c1 c2 a1.sinks=k1 k2 2描述source a1.source.r1.type=exec a1.source.r1.command=tail -F /opt/module/flum
Flume进阶之复制和多路复用、负载均衡和故障转移、聚合案例
花&败
10-07 451
hadoop102 上的Flume-1 监控文件/opt/module/flume-1.9.0/group.log，hadoop103 上的Flume-2 监控某一个端口的数据流，Flume-1 与 Flume-2 将数据发送给 hadoop104 上的 Flume-3，Flume-3 将最终数据打印到控制台。故障转移需求：使用 Flume1 监控一个端口，其 sink 组中的 sink 分别对接 Flume2 和 Flume3，采用FailoverSinkProcessor，实现故障转移的功能。
评论
添加红包

请填写红包祝福语或标题

个

红包个数最小为10个

元

红包金额最低5元

当前余额3.43元 前往充值 >
需支付：10.00元
成就一亿技术人!
领取后你会自动成为博主和红包主的粉丝 规则
hope_wisdom
发出的红包

打赏作者

不太灵光的程序员

有用的话可以请博主喝杯咖啡续命

¥1 ¥2 ¥4 ¥6 ¥10 ¥20
扫码支付：¥1
获取中
扫码支付

您的余额不足，请更换扫码支付或充值

打赏作者

实付元
使用余额支付
点击重新获取
扫码支付
钱包余额 0

抵扣说明：

1.余额是钱包充值的虚拟货币，按照1:1的比例进行支付金额的抵扣。
2.余额无法直接购买下载，可以购买VIP、付费专栏及课程。

余额充值