flume案例三：扇入

扇入(fan in)-Flume与Flume之间数据传递，多Flume汇总数据到单Flume

**目标：**agent1 监控.log文件 agent2 监控某一个端口的数据流
汇总数据发送给agent3 并写道hdfs上

#agent1  监控.log文件
a1.sources=s1
a1.sinks=k1
a1.channels=c1#设置sources
a1.sources.s1.type=spooldir
a1.sources.s1.spoolDir=/root/data/flumeFile/test
a1.sources.s1.includePattern=tmp_[0-9]{4}-[0-9]{2}-[0-9]{2}.log
a1.sources.s1.deserializer=LINE
a1.sources.s1.deserializer.maxLineLength=60000
#设置channel
a1.channels.c1.type=memory
a1.channels.c1.capacity=1000
a1.channels.c1.transactionCapacity=100
#设置sink
a1.sinks.k1.type=avro
a1.sinks.k1.hostname=localhost
a1.sinks.k1.port=5555
#设置连接
a1.sources.s1.channels=c1
a1.sinks.k1.channels=c1

#agent2  监控某一个端口的数据流
a2.sources= s1
a2.sinks =k1
a2.channels =c1
#设置sources
a2.sources.s1.type=netcat
a2.sources.s1.hostname=localhost
a2.sources.s1.port=1111
#设置channel
a2.channels.c1.type=memory
a2.channels.c1.capacity=1000
a2.channels.c1.transactionCapacity=100
#设置sinks
a2.sinks.k1.type=avro
a2.sinks.k1.hostname=localhost
a2.sinks.k1.port=5555
#设置连接
a2.sources.s1.channels=c1
a2.sinks.k1.channels=c1

#agent3 监听avro端口 汇总外部来的avro客户端的数据，并写道hdfs上
a3.sources= s1
a3.sinks =k1
a3.channels =c1
#设置sources
a3.sources.s1.type=avro
a3.sources.s1.hostname=localhost
a3.sources.s1.port=5555
#设置channel
a3.channels.c1.type=memory
a3.channels.c1.capacity=1000
a3.channels.c1.transactionCapacity=100
#设置sink
a3.sinks.k1.type=hdfs
a3.sinks.k1.hdfs.fileType=DataStream
a3.sinks.k1.hdfs.filePrefix=tmp
a3.sinks.k1.hdfs.fileSuffix=.csv
a3.sinks.k1.hdfs.path=hdfs://192.168.241.132:9000/flume/%Y-%m-%d
a3.sinks.k1.hdfs.useLocalTimeStamp=true
a3.sinks.k1.hdfs.batchSize=640
a3.sinks.k1.hdfs.rollCount= 0
a3.sinks.k1.hdfs.rollSize=6400000
a3.sinks.k1.hdfs.rollInterval =30
#设置连接
a3.sources.s1.channels=c1
a3.sinks.k1.channel=c1

ps:agent1 和agent2中的sink avro 端口号要保持一致以及agent的source端口也要一样
启动顺序：a2和a1 随意但要先启动a3

$ bin/flume-ng agent --conf conf/ --name a3 --conf-file jobconf/flume33.conf$ bin/flume-ng agent --conf conf/ --name a2 --conf-file jobconf/flume22.conf$ bin/flume-ng agent --conf conf/ --name a1 --conf-file jobconf/flume11.conf

flume案例三：扇入相关推荐

第1节 flume：15、flume案例二，通过自定义拦截器实现数据的脱敏
1.7.flume案例二案例需求: 在数据采集之后,通过flume的拦截器,实现不需要的数据过滤掉,并将指定的第一个字段进行加密,加密之后再往hdfs上面保存原始数据与处理之后的数据对比图一 ...
2021年大数据Flink（三十六）：Table与SQL 案例三
目录案例三需求编码步骤代码实现-方式1 代码实现-方式2 案例三需求使用Flink SQL来统计5秒内每个用户的订单总数.订单的最大金额.订单的最小金额也就是每隔5秒统计最近5秒的每 ...
2021年大数据Flink（二十一）：案例三会话窗口
目录案例三会话窗口需求代码实现案例三会话窗口需求设置会话超时时间为10s,10s内没有数据到来,则触发上个窗口的计算代码实现 package cn.it.window;import ...
数据分析与挖掘 - R语言：贝叶斯分类算法（案例三）
案例三比较简单,不需要自己写公式算法,使用了R自带的naiveBayes函数. 代码如下: > library(e1071)> classifier<-naiveBayes(iris ...
C++多态案例三-电脑组装
C++多态案例三-电脑组装多态案例三-电脑组装多态案例三-电脑组装案例描述: 电脑主要组成部件为 CPU(用于计算),显卡(用于显示),内存条(用于存储) 将每个零件封装出抽象基类,并且提供不同 ...
Redis简单案例(三) 连续登陆活动的简单实现
原文:Redis简单案例(三) 连续登陆活动的简单实现连续登陆活动,或许大家都不会陌生,简单理解就是用户连续登陆了多少天之后,系统就会送一些礼品给相应的用户.最常见的莫过于游戏和商城这些.游戏就送 ...
案例三：执行 JavaScript 语句
案例三:执行 JavaScript 语句隐藏百度图片 from selenium import webdriverdriver = webdriver.PhantomJS() driver.get( ...
Flume案例Ganglia监控
Flume案例和Flume监控系统的使用: 安装将apache-flume-1.7.0-bin.tar.gz上传到linux的/opt/software目录下解压apache-flume-1.7. ...
zabbix生产环境案例(三)
生产环境案例(三) 链接:https://pan.baidu.com/s/1q5YwJMTcZLcS5OQ0iOu44A 提取码:8gdi 复制这段内容后打开百度网盘手机App,操作更方便哦 1. Z ...

flume案例三：扇入

扇入(fan in)-Flume与Flume之间数据传递，多Flume汇总数据到单Flume

flume案例三：扇入相关推荐

最新文章

热门文章