Flume 简单使用（avro、exec、spooldir、hdfs）

81 阅读 0 评论 54 点赞

我是靠谱客的博主儒雅服饰，最近开发中收集的这篇文章主要介绍Flume 简单使用（avro、exec、spooldir、hdfs），觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

需求一：

监听文件内容变动，将新增内容输出到控制台。

#指定agent的sources,sinks,channels
a1.sources = s1  
a1.sinks = k1  
a1.channels = c1  
   
#配置sources属性
a1.sources.s1.type = exec
a1.sources.s1.command = tail -F /tmp/log.txt
a1.sources.s1.shell = /bin/bash -c
  
#配置channel类型
a1.channels.c1.type = memory
  
#配置sink 
a1.sinks.k1.type = logger

#将sources与channels进行绑定
a1.sources.s1.channels = c1

#将sinks与channels进行绑定  
a1.sinks.k1.channel = c1

案例二：

监听指定目录，将目录下新增文件上传到 hdfs。

#指定agent的sources,sinks,channels
a1.sources = s1  
a1.sinks = k1  
a1.channels = c1  
   
#配置sources属性
a1.sources.s1.type =spooldir  
a1.sources.s1.spoolDir =/tmp/logs
a1.sources.s1.basenameHeader = true
a1.sources.s1.basenameHeaderKey = fileName 

#配置channel类型
a1.channels.c1.type = memory
   
#配置sink 
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H/
a1.sinks.k1.hdfs.filePrefix = %{fileName}
#生成的文件类型，默认是Sequencefile，可用DataStream，则为普通文本
a1.sinks.k1.hdfs.fileType = DataStream  
a1.sinks.k1.hdfs.useLocalTimeStamp = true

#将sources与channels进行绑定  
a1.sources.s1.channels =c1 

#将sinks与channels进行绑定  
a1.sinks.k1.channel = c1

案例三：

监听文件变化，发送到指定端口。

#指定agent的sources,sinks,channels
a1.sources = s1
a1.sinks = k1
a1.channels = c1

#配置sources属性
a1.sources.s1.type = exec
a1.sources.s1.command = tail -F /tmp/log.txt
a1.sources.s1.shell = /bin/bash -c
a1.sources.s1.channels = c1

#配置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 = daidai
a1.sinks.k1.port = 8888
a1.sinks.k1.batch-size = 1
a1.sinks.k1.channel = c1

案例四：

监听端口数据，发送到控制台。

#指定agent的sources,sinks,channels
a2.sources = s2
a2.sinks = k2
a2.channels = c2

#配置sources属性
a2.sources.s2.type = avro
a2.sources.s2.bind = daidai
a2.sources.s2.port = 8888

#配置channel类型
a2.channels.c2.type = memory
a2.channels.c2.capacity = 1000
a2.channels.c2.transactionCapacity = 100

#配置sink
a2.sinks.k2.type = logger

#将sources与channels进行绑定
a2.sources.s2.channels = c2
#将sinks与channels进行绑定
a2.sinks.k2.channel = c2