返回 导航

大数据

hangge.com

Flume日志采集工具使用详解8（样例2：Channel选择器、多Channel）

作者：hangge | 2024-03-15 08:40
    一个 Source 支持给多个 Channel 发送数据，如果 source 后面接了多个 channel，到底是给所有的 channel 都发，还是根据规则发送到不同 channel，这些是由 Channel Selectors 来控制的。
    Channel Selectors 类型包括：Replicating Channel Selector 和 Multiplexing Channel Selector
  • Replicating Channel Selector 是默认的 channel 选择器，它会将 Source 采集过来的 Event 发往所有 Channel
  • Multiplexing Channel Selector 表示会根据 Event 中 header 里面的值将 Event 发往不同的 Channel

八、Channel 选择器的使用1：Replicating Channel Selector

1，整体架构

（1）在这个案例中我们使用 Replicating 选择器，将 source 采集到的数据重复发送给两个 channel，最后每个 channel 后面接一个 sink，负责把数据存储到不同存储介质中，方便后期使用。

（2）在实际工作中这种需求还是比较常见的，就是我们希望把一份数据采集过来以后，分别存储到不同的存储介质中，不同存储介质的特点和应用场景是不一样的，典型的就是 hdfssink 和 kafkasink：
  • 通过 hdfssink 实现离线数据落盘存储，方便后面进行离线数据计算。
  • 通过 kafkasink 实现实时数据存储，方便后面进行实时计算。

2，配置 Agent

（1）启动 Flume 任务其实就是启动一个 Agent。首先我们进入 flume 程序包中的 conf 文件夹，创建一个 Agent 配置文件：
cd conf
vi example.conf

（2）接着在配置文件中添加如下内容。 
# 定义 Agent 内部 3 大组件的名称
a1.sources = r1
a1.sinks = k1 k2
a1.channels =c1 c2
 
# 配置 Source 组件
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 6666
 
# 配置 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://node1: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.k1.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

3，测试 Agent

（1）首先我们执行如下命令启动此 Agent： 
nohup bin/flume-ng agent --name a1 --conf conf --conf-file conf/example.conf &

（2）接着打开一个新的终端窗口首先使用 telnet 连接配置的 6666 端口：
telnet 192.168.60.9 6666

（3）连接上后发送两条数据：

（4）查看 Flume 的日志文件，可以看到里面有相关数据的日志信息：

（5）查看 hdfs 中生成的文件中的内容也是正常，说明多 Channel 配置成功生效了。

九、Channel 选择器的使用2：Multiplexing Channel Selector

1，整体架构

    在这个案例中我们使用 Multiplexing 选择器，将 source 采集到的数据按照一定规则发送给两个 channel，最终再把不同 channel 中的数据存储到不同介质中。

2，配置 Agent

（1）启动 Flume 任务其实就是启动一个 Agent。首先我们进入 flume 程序包中的 conf 文件夹，创建一个 Agent 配置文件：
cd conf
vi example.conf

（2）接着在配置文件中添加如下内容。
提示：在这里面我们需要用到正则抽取拦截器在 Event 的 header 中生成 key-value 作为 Multiplexing 选择器的规则
# 定义 Agent 内部 3 大组件的名称
a1.sources = r1
a1.sinks = k1 k2
a1.channels =c1 c2
 
# 配置 Source 组件
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 6666

# 配置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://node1: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.k1.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

3，测试 Agent

（1）首先我们执行如下命令启动此 Agent： 
nohup bin/flume-ng agent --name a1 --conf conf --conf-file conf/example.conf &

（2）接着打开一个新的终端窗口首先使用 telnet 连接配置的 6666 端口：
telnet 192.168.60.9 6666

（3）连接上后发送两条数据： 
{"name":"jack","age":19,"city":"bj"}
{"name":"tom","age":26,"city":"sh"}

（4）查看 Flume 的日志文件，可以看到里面只有 city 为 bj 的数据信息：

（5）查看 hdfs 中生成的文件，里面只有 city 不为 bj 的数据，说明多 Channel 选择器配置成功生效了。
评论

全部评论（0）

回到顶部