Flume使用案例之Flume与Flume之间数据传递,多Flume汇总数据到单Flume

楼高2年前技术文章519

目标:flume11监控文件hive.logflume-22监控某一个端口的数据流,flume11flume-22将数据发送给flume-33flume33将最终数据写入到HDFS

分步实现:

1. 创建flume11.conf,用于监控hive.log文件,同时sink数据到flume-33

# 1 agent

a1.sources = r1

a1.sinks = k1

a1.channels = c1

 

# 2 source

a1.sources.r1.type = exec

a1.sources.r1.command = tail -F /opt/Andy

a1.sources.r1.shell = /bin/bash -c

 

# 3 sink

a1.sinks.k1.type = avro

a1.sinks.k1.hostname = dtstack_hdfs

a1.sinks.k1.port = 4141

 

# 4 channel

a1.channels.c1.type = memory

a1.channels.c1.capacity = 1000

a1.channels.c1.transactionCapacity = 100

 

# 5. Bind

a1.sources.r1.channels = c1

a1.sinks.k1.channel = c1

 2. 创建flume-22.conf,用于监控端口44444数据流,同时sink数据到flume-33

# 1 agent

a2.sources = r1

a2.sinks = k1

a2.channels = c1

 

# 2 source

a2.sources.r1.type = netcat

a2.sources.r1.bind = dtstack_hdfs

a2.sources.r1.port = 44444

 

#3 sink

a2.sinks.k1.type = avro

a2.sinks.k1.hostname = dtstack_hdfs

a2.sinks.k1.port = 4141

 

# 4 channel

a2.channels.c1.type = memory

a2.channels.c1.capacity = 1000

a2.channels.c1.transactionCapacity = 100

 

# 5 Bind

a2.sources.r1.channels = c1

a2.sinks.k1.channel = c1

 3.  创建flume33.conf,用于接收flume11flume22发送过来的数据流,最终合并后sinkHDFS

# 1 agent

a3.sources = r1

a3.sinks = k1

a3.channels = c1

 

# 2 source

a3.sources.r1.type = avro

a3.sources.r1.bind = dtstack_hdfs

a3.sources.r1.port = 4141

 

# 3 sink

a3.sinks.k1.type = hdfs

a3.sinks.k1.hdfs.path = hdfs://dtstack_hdfs:9000/flume3/%H

#上传文件的前缀

a3.sinks.k1.hdfs.filePrefix = flume3-

#是否按照时间滚动文件夹

a3.sinks.k1.hdfs.round = true

#多少时间单位创建一个新的文件夹

a3.sinks.k1.hdfs.roundValue = 1

#重新定义时间单位

a3.sinks.k1.hdfs.roundUnit = hour

#是否使用本地时间戳

a3.sinks.k1.hdfs.useLocalTimeStamp = true

#积攒多少个EventflushHDFS一次

a3.sinks.k1.hdfs.batchSize = 100

#设置文件类型,可支持压缩

a3.sinks.k1.hdfs.fileType = DataStream

#多久生成一个新的文件

a3.sinks.k1.hdfs.rollInterval = 600

#设置每个文件的滚动大小大概是128M

a3.sinks.k1.hdfs.rollSize = 134217700

#文件的滚动与Event数量无关

a3.sinks.k1.hdfs.rollCount = 0

#最小冗余数

a3.sinks.k1.hdfs.minBlockReplicas = 1

 

# 4 channel

a3.channels.c1.type = memory

a3.channels.c1.capacity = 1000

a3.channels.c1.transactionCapacity = 100

 

# 5 Bind

a3.sources.r1.channels = c1

a3.sinks.k1.channel = c1

 4. 执行测试:分别开启对应flume-job(依次启动flume-33flume-22flume11),同时产生文件变动并观察结果

$ 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

数据发送

1) telnet dtstack_hdfs 44444    打开后发送java

2) /opt/Andy 中追加python

相关文章

pg_probackup

一、pg_probackup概述pg_probackup 是一款免费的postgres数据库集群备份工具,与其他备份工具相比,它主要有如下一些优势:提供增量备份,增量备份一定程度上可以节省磁盘空间的使...

Golang 垃圾回收

Golang 垃圾回收

1、标记清除算法Golang 使用标记清除算法作为垃圾回收器的一部分。标记清除算法是一种常见的垃圾回收算法,它通过标记和清除未被引用的对象来回收内存空间。Golang 中,垃圾回收器会定期扫描堆空间,...

ORA-00054报错处理

执行语句报错:ORA-00054: resource busy and acquire with NOWAIT specified or timeout expired原因:如错误信息提示,资源正忙(...

MySQL运维实战(4.1) MySQL表存储引擎

MySQL表的特点和其他数据库如Oracle、SQL Server相比,mysql有一些特点:MySQL使用插件式存储引擎,同一个数据库中的表可以使用不同的存储引擎。存储引擎决定了表的物理存储格式。表...

Atlas集成Hive

Atlas集成Hive

1 集成原理2 验证Hive元数据采集效果(1) 查看Atlas里是否有Hive元数据(2) 进入Hive创建一个库表create database if not exists foo;(3) 进入A...

A集群导入B集群中的高可用版rancher,一直处于pending状态

A集群导入B集群中的高可用版rancher,一直处于pending状态

问题现象:已知在B集群中采用helm方式部署了一个高可用版本的rancher,该rancher中已经配置导入了三套集群,并且三套集群状态在rancher控制台处均显示正常,日常可借助该rancher管...

发表评论    

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。