AI 助理
备案控制台
开发者社区 云计算 文章 正文

bigdata-14-Flume高级组件

2024-03-28 63
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： bigdata-14-Flume高级组件

3.4 Flume高级组件

高级组件

  • Source Interceptors：Source可以指定一个或者多个拦截器按先后顺序依次对采集到的数据进行处 理。
  • Channel Selectors：Source发往多个Channel的策略设置，如果source后面接了多个channel，到 底是给所有的channel都发，还是根据规则发送到不同channel，这些是由Channel Selectors来控制 的
  • Sink Processors：Sink 发送数据的策略设置，一个channel后面可以接多个sink，channel中的数据 是被哪个sink获取，这个是由Sink Processors控制的

事件

Event是Flume传输数据的基本单位，也是事务的基本单位，在文本文件中，通常一行记录就是一个Event

Event中包含header和body；

  • body是采集到的那一行记录的原始内容
  • header类型为Map，里面可以存储一些属性信息，方便后面使用
  • 我们可以在Source中给每一条数据的header中增加key-value，然后在Channel和Sink中使用header的值

Source Interceptors

常见的有：Timestamp Interceptor、Host Interceptor、Search and Replace Interceptor 、Static Interceptor、Regex Extractor Interceptor

  • Timestamp Interceptor：向event中的header里面添加timestamp 时间戳信息
  • Host Interceptor：向event中的header里面添加host属性，host的值为当前机器的主机名或者ip
  • Search and Replace Interceptor：根据指定的规则查询Event中body里面的数据，然后进行替换， 这个拦截器会修改event中body的值，也就是会修改原始采集到的数据内容
  • Static Interceptor：向event中的header里面添加固定的key和value
  • Regex Extractor Interceptor：根据指定的规则从Event中的body里面抽取数据，生成key和value， 再把key和value添加到header中

我们event中的header里面添加key-value类型的数据，方便后面的channel和sink组件使用，对采集到的原始数据内容没有任何影响。

Search and Replace Interceptor是会根据规则修改event中body里面的原始数据内容，对header没有任 何影响，使用这个拦截器需要特别小心，因为他会修改原始数据内容。

案例数据准备

video_info
{"id":"14943445328940974601","uid":"840717325115457536","lat":"53.530598","lnt":"-2.5620373","hots":0,"title":"0","status":"1","topicId":"0","end_time":"1494344570","watch_num":0,"share_num":"1","replay_url":null,"replay_num":0,"start_time":"1494344544","timestamp":1494344571,"type":"video_info"}
user_info
{"uid":"861848974414839801","nickname":"mick","usign":"","sex":1,"birthday":"","face":"","big_face":"","email":"abc@qq.com","mobile":"","reg_type":"102","last_login_time":"1494344580","reg_time":"1494344580","last_update_time":"1494344580","status":"5","is_verified":"0","verified_info":"","is_seller":"0","level":1,"exp":0,"anchor_level":0,"anchor_exp":0,"os":"android","timestamp":1494344580,"type":"user_info"}
gift_record
{"send_id":"834688818270961664","good_id":"223","video_id":"14943443045138661356","gold":"10","timestamp":1494344574,"type":"gift_record"}

首先确定这是一份json格式的数据

包括：视频信息、用户信息、送礼信息

当业务平台运行时，会实时产生这些日志数据，我们希望把这些数据采集到hdfs上进行分目录存储，并且根据type类型来进行区分。

针对这个需求配置agent的话，source使用基于文件的execsource、channle使用基于文件的channel， 我们希望保证数据的完整性和准确性，sink使用hdfssink。

最终要生成的目录如下

hdfs://192.168.197.101:9000/moreType/20200101/videoInfo
hdfs://192.168.197.101:9000/moreType/20200101/userInfo
hdfs://192.168.197.101:9000/moreType/20200101/giftRecord

这里就要思考一个问题，如何获取数据类型？

答案：通过source的拦截器可以向event的header中添加key-value，然后在 后面的channle或者sink中获取key-value的值

那我们在这就可以通过Regex Extractor Interceptor获取原始数据中的type字段的值，获取出来以后存储 到header中，这样在sink阶段就可以获取到了。

但是这个时候直接获取到的type的值是不满足要求的，需要对type的值进行转换，去掉下划线，转化为驼峰命名形式。

所以最终可以先使用Search and Replace Interceptor对原始数据中type的值进行转换，然后使用Regex Extractor Interceptor指定规则获取type字段的值，添加到header中。

最终流程：

Exec Source -> Search and Replace Interceptor->Regex Extractor Interceptor -> file channel -> hdfs sink

在bigdata04机器上创建 file-to-hdfs-moreType.conf

配置文件如下:

# agent的名称是a1
# 指定source组件、channel组件和Sink组件的名称
a1.sources = r1
a1.channels = c1
a1.sinks = k1
# 配置source组件
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /data/log/moreType.log
# 配置拦截器 [多个拦截器按照顺序依次执行]
a1.sources.r1.interceptors = i1 i2 i3 i4
a1.sources.r1.interceptors.i1.type = search_replace
a1.sources.r1.interceptors.i1.searchPattern = "type":"video_info"
a1.sources.r1.interceptors.i1.replaceString = "type":"videoInfo"
a1.sources.r1.interceptors.i2.type = search_replace
a1.sources.r1.interceptors.i2.searchPattern = "type":"user_info"
a1.sources.r1.interceptors.i2.replaceString = "type":"userInfo"
a1.sources.r1.interceptors.i3.type = search_replace
a1.sources.r1.interceptors.i3.searchPattern = "type":"gift_record"
a1.sources.r1.interceptors.i3.replaceString = "type":"giftRecord"
a1.sources.r1.interceptors.i4.type = regex_extractor
a1.sources.r1.interceptors.i4.regex = "type":"(\\w+)"
a1.sources.r1.interceptors.i4.serializers = s1
a1.sources.r1.interceptors.i4.serializers.s1.name = logType
# 配置channel组件
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /data/soft/apache-flume-1.9.0-bin/data/moreType/checkpoint
a1.channels.c1.dataDirs = /data/soft/apache-flume-1.9.0-bin/data/moreType/data
# 配置sink组件
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://192.168.197.100:9000/moreType/%Y%m%d/%{logType}
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.rollInterval = 3600
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#增加文件前缀和后缀
a1.sinks.k1.hdfs.filePrefix = data
a1.sinks.k1.hdfs.fileSuffix = .log
# 把组件连接起来
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

注意：这里面的拦截器，拦截器可以设置一个或者多个，source采集的每一条数据都会经过所有 的拦截器进行处理，多个拦截器按照顺序执行。

后面也统一设置了hdfs文件的前缀和后缀。

首先我们先将测试数据放入指定目录

cd /data/log
vi moreType.log

上传配置文件到flume conf文件夹下

启动bigdata04 Agent并去bigdata01查看结果

启动命令如下：

../bin/flume-ng agent --name a1 --conf /data/soft/apache-flume-1.9.0-bin/conf/ --conf-file file-to-hdfs-moreType.conf -Dflume.root.logger=INFO,console

成功采集数据

测试成功

Channel Selectors

Channel Selectors类型包括：Replicating Channel Selector 和Multiplexing Channel Selector

官网案例如下：

简单讲：这个案例的意思就是source数据会分发到c1、c2、c3三个channel，flume可以保证c1、c2收到数据，但是c3是否能够收到数据并不保证。

selector.optional参数是一个可选项，可以不用配置

注意这个channel位置是复数，就表明了可以配置多个channel

此外还有Multiplexing Channel Selector，它表示会根据Event中header里面的值将Event发往不同的Channel，官网案例如下

在这个例子的配置中，指定了4个channel,c1、c2、c3、c4 source采集到的数据具体会发送到哪个channel中，会根据event中header里面的state属性的值，这个是通过selector.header控制的

  • 如果state属性的值是CZ，则发送给c1
  • 如果state属性的值是US，则发送给c2 c3
  • 如果state属性的值是其它值，则发送给c4

这些规则是通过selector.mapping和selector.default控制的

案例1

多Channel之Replicating Channel Selector

组件关系图如下：

这张场景还是比较常见的，分析如下：

1.通过Replicating选择器将Source数据分发到两个channel。

2.将不同channel数据分发到两个不同的sink端进行消费。

实际工作中：

一般对接hdfs+kafka。

具体操作如下：

1.在bigdata04增加配置文件 tcp-to-replicatingchannel.conf

内容如下：

# agent的名称是a1
# 指定source组件、channel组件和Sink组件的名称
a1.sources = r1
a1.channels = c1 c2
a1.sinks = k1 k2
# 配置source组件
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444
# 配置channle选择器[默认就是replicating，所以可以省略]
a1.sources.r1.selector.type = replicating
# 配置channel组件
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100
# 配置sink组件
a1.sinks.k1.type = logger
a1.sinks.k2.type = hdfs
a1.sinks.k2.hdfs.path = hdfs://192.168.197.101:9000/replicating
a1.sinks.k2.hdfs.fileType = DataStream
a1.sinks.k2.hdfs.writeFormat = Text
a1.sinks.k2.hdfs.rollInterval = 3600
a1.sinks.k2.hdfs.rollSize = 134217728
a1.sinks.k2.hdfs.rollCount = 0
a1.sinks.k2.hdfs.useLocalTimeStamp = true
a1.sinks.k2.hdfs.filePrefix = data
a1.sinks.k2.hdfs.fileSuffix = .log
# 把组件连接起来
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2

这里使用netcat方式方便生成测试数据

2.启动bigdata04 Agent

../bin/flume-ng agent --name a1 --conf /data/soft/apache-flume-1.9.0-bin/conf/ --conf-file tcp-to-replicatingchannel.conf -Dflume.root.logger=INFO,console

3.通过telnet连接到socket 生产测试数据

可以看到我们既在控制台输出了数据，由存储了hdfs数据文件

案例2

Multiplexing Channel Selector

组件关系图如下：

测试数据如下：

{"name":"jack","age":19,"city":"bj"}
{"name":"tom","age":26,"city":"sh"}

配置文件内容如下：

在bigdata04上创建新文件 tcp-to-multiplexingchannel.conf

# agent的名称是a1
# 指定source组件、channel组件和Sink组件的名称
a1.sources = r1
a1.channels = c1 c2
a1.sinks = k1 k2
# 配置source组件
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444
# 配置source拦截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = regex_extractor
a1.sources.r1.interceptors.i1.regex = "city":"(\\w+)"
a1.sources.r1.interceptors.i1.serializers = s1
a1.sources.r1.interceptors.i1.serializers.s1.name = city
# 配置channle选择器
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = city
a1.sources.r1.selector.mapping.bj = c1
a1.sources.r1.selector.default = c2
# 配置channel组件
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100
# 配置sink组件
a1.sinks.k1.type = logger
a1.sinks.k2.type = hdfs
a1.sinks.k2.hdfs.path = hdfs://192.168.197.101:9000/multiplexing
a1.sinks.k2.hdfs.fileType = DataStream
a1.sinks.k2.hdfs.writeFormat = Text
a1.sinks.k2.hdfs.rollInterval = 3600
a1.sinks.k2.hdfs.rollSize = 134217728
a1.sinks.k2.hdfs.rollCount = 0
a1.sinks.k2.hdfs.useLocalTimeStamp = true
a1.sinks.k2.hdfs.filePrefix = data
a1.sinks.k2.hdfs.fileSuffix = .log
# 把组件连接起来
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2

启动flume agent

../bin/flume-ng agent --name a1 --conf /data/soft/apache-flume-1.9.0-bin/conf/ --conf-file tcp-to-multiplexingchannel.conf -Dflume.root.logger=INFO,console

ctrl + ]退出telnet

发送测试数据

数据采集成功

成功过滤出city是北京的数据

这样就实现了，根据规则把source采集到的数据分发到不同channel中，最终输出到不同存储介质中。 这就是Multiplexing Channel Selector的应用了

Sink Processors

  • Default Sink Processor：默认的，不用配置sinkgroup，就是咱们现在使用的这种最普通的形式，一个 channel后面接一个sink的形式
  • Load balancing Sink Processor是负载均衡处理器，一个channel后面可以接多个sink，这多个sink属于 一个sink group，根据指定的算法进行轮询或者随机发送，减轻单个sink的压力。
  • Failover Sink Processor是故障转移处理器，一个channle后面可以接多个sink，这多个sink属于一个 sink group，按照sink的优先级，默认先让优先级高的sink来处理数据。

负载均衡

组件关系图

processor.sinks：指定这个sink groups中有哪些sink

processor.type：针对负载均衡的sink处理器，这里需要指定load_balance

processor.selector：此参数的值内置支持两个，round_robin和random，round_robin表示轮询，按照 sink的顺序，轮流处理数据，random表示随机

processor.backoff：默认为false，设置为true后，故障的节点会列入黑名单，过一定时间才会再次发送 数据，如果还失败，则等待时间是指数级增长；一直到达到最大的时间。

processor.selector.maxTimeOut：最大的黑名单时间，默认是30秒

故障转移

参考配置：

# agent的名称是a1
# 指定source组件、channel组件和Sink组件的名称
a1.sources = r1
a1.channels = c1
a1.sinks = k1 k2
# 配置source组件
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444
# 配置channel组件
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = 192.168.197.102
a1.sinks.k1.port = 41414
a1.sinks.k1.batch-size = 1
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = 192.168.197.103
a1.sinks.k2.port = 41414
a1.sinks.k2.batch-size = 1
# 配置sink策略
a1.sinkgroups = g1
a1.sinkgroups.g1.sinks = k1 k2
a1.sinkgroups.g1.processor.type = load_balance
a1.sinkgroups.g1.processor.backoff = true
a1.sinkgroups.g1.processor.selector = round_robin
# 把组件连接起来
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c1

processor.type：针对故障转移的sink处理器，使用

failover processor.priority.：指定sink group中每一个sink组件的优先级，默认情况下channel中的数据会被 优先级比较高的sink取走

processor.maxpenalty：sink发生故障之后，最大等待时间

参考配置文件：

# agent的名称是a1
# 指定source组件、channel组件和Sink组件的名称
a1.sources = r1
a1.channels = c1
a1.sinks = k1 k2
# 配置source组件
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444
# 配置channel组件
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = 192.168.197.102
a1.sinks.k1.port = 41414
a1.sinks.k1.batch-size = 1
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = 192.168.182.102
a1.sinks.k2.port = 41414
a1.sinks.k2.batch-size = 1
# 配置sink策略
a1.sinkgroups = g1
a1.sinkgroups.g1.sinks = k1 k2
a1.sinkgroups.g1.processor.type = failover
a1.sinkgroups.g1.processor.priority.k1 = 5
a1.sinkgroups.g1.processor.priority.k2 = 10
a1.sinkgroups.g1.processor.maxpenalty = 10000
# 把组件连接起来
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c1
文章标签：
存储
负载均衡
数据采集
算法
JSON
关键词：
Flume组件
Cool架构
目录
相关文章
Cool架构
|
7月前
bigdata-12-Flume核心组件
bigdata-12-Flume核心组件
Cool架构
65 0
土木林森
|
4月前
|
数据采集 存储 Apache
Flume核心组件大揭秘：Agent、Source、Channel、Sink，一文掌握数据采集精髓！
【8月更文挑战第24天】Flume是Apache旗下的一款顶级服务工具，专为大规模日志数据的收集、聚合与传输而设计。其架构基于几个核心组件：Agent、Source、Channel及Sink。Agent作为基础执行单元，整合Source（数据采集）、Channel（数据暂存）与Sink（数据传输）。本文通过实例深入剖析各组件功能与配置，包括Avro、Exec及Spooling Directory等多种Source类型，Memory与File Channel方案以及HDFS、Avro和Logger等Sink选项，旨在提供全面的Flume应用指南。
土木林森
212 1
爱吃糖的范同学
|
7月前
|
存储 消息中间件 Kafka
【Flume】Flume 核心组件分析
【4月更文挑战第4天】【Flume】Flume 核心组件分析
爱吃糖的范同学
159 7
北惜
|
数据采集 消息中间件 监控
大数据组件-Flume集群环境搭建
大数据组件-Flume集群环境搭建
北惜
205 0
北惜
|
Oracle 大数据 关系型数据库
大数据组件-Flume集群环境的启动与验证
大数据组件-Flume集群环境的启动与验证
北惜
142 0
安然AR
|
存储 数据采集 消息中间件
大数据数据采集的数据采集（收集/聚合）的Flume之基本组件的Sink：从Channel中取数据
在Flume中，Sink是数据采集和传输过程中的最终组件。它负责从Channel缓冲区中获取数据并将其存储到目标存储系统中。
安然AR
266 0
安然AR
|
存储 数据采集 缓存
大数据数据采集的数据采集（收集/聚合）的Flume之基本组件的Channel：临时存储数据的管道
在Flume中，Channel是数据采集和传输过程中的一个重要组件。它负责存储从Source获取的数据，并将其转发给Sink进行处理和存储。
安然AR
152 0
安然AR
|
数据采集 消息中间件 存储
大数据数据采集的数据采集（收集/聚合）的Flume之基本组件的Event：数据基本单元
在Flume中，Event是数据采集和传输过程中的基本单元。每个Event都代表了一个数据记录，包括一个头信息和一个消息体。
安然AR
224 0
安然AR
|
存储 数据采集 JSON
大数据数据采集的数据采集（收集/聚合）的Flume之基本组件的Source：数据的收集端
在Flume中，Source是数据采集和传输过程中的一个重要组件。它负责从生产者获取数据并将其发送到Channel缓冲区中，为后续的数据处理和存储提供支持。
安然AR
202 0
安然AR
|
存储 数据采集 监控
大数据数据采集的数据采集（收集/聚合）的Flume之基本组件的Agent
在Flume中，Agent是数据采集和传输过程中的核心组件。它负责从Source获取数据，并将其发送到Channel缓冲区中，最后将经过处理的数据发送给Sink进行存储。
安然AR
416 0

热门文章

最新文章

  • 1
    Flume NG 简介及配置实战
  • 2
    flume java介绍
  • 3
    模拟使用Flume监听日志变化，并且把增量的日志文件写入到hdfs中
  • 4
    使用Flume将日志导入OSS
  • 5
    【大数据新手上路】“零基础”系列课程--Flume收集网站日志数据到MaxCompute
  • 6
    flume与kafka整合
  • 7
    flume详细介绍，安装，配置
  • 8
    08【在线日志分析】之Flume Agent(聚合节点) sink to kafka cluster
  • 9
    flume1.6 centos6.x 安装与配置
  • 10
    【Flume】HDFSSink源码理解
  • 1
    使用 Flume 将 CSV 数据导入 Kafka：实现实时数据流
    146
  • 2
    Flume（二）【Flume 进阶使用】（4）
    53
  • 3
    Flume（二）【Flume 进阶使用】（3）
    40
  • 4
    Flume（二）【Flume 进阶使用】（2）
    45
  • 5
    Flume（二）【Flume 进阶使用】（1）
    67
  • 6
    Flume 快速入门【概述、安装、拦截器】
    1190
  • 7
    Flume 配置文件编写技巧（包会的，抄就完了）
    386
  • 8
    Flume 拦截器概念及自定义拦截器的运用
    218
  • 9
    Flume实时读取本地/目录文件到HDFS
    206
  • 10
    【数据采集与预处理】流数据采集工具Flume
    208

    • 相关课程

    更多
  • Flume基础应用实战 - 企业全场景解决方案
  • 数据采集系统 Flume 快速入门
  • Hadoop快速入门
  • Apache Flink 入门
  • HBase入门教程
  • Hadoop 分布式计算框架 MapReduce

    • 相关电子书

    更多
  • Dataflow with Apache NiFi
  • The Avant-garde of Apache NiFi
  • Apache NiFi 1.0 in Nutshell

    • 相关实验场景

    更多
  • 搭建Hadoop环境
    • 下一篇
    阿里云对象存储OSS收费标准：存储、流量和请求等多个计费项详解