flume+kafka+hdfs详解

  1. 云栖社区>
  2. 博客列表>
  3. 正文

flume+kafka+hdfs详解

神巧合 2016-08-28 14:04:00 浏览1183 评论0

摘要: flume架构图 单节点flume配置 flume-1.4.0  启动flume bin/flume-ng agent --conf ./conf  -f conf/flume-conf.properties -Dflume.root.logger=DEBUG,console -n agent -n表示配置文件中agent的名字 agent.sources = r1 agent.sinks 

flume架构图

wKiom1YPrdjguqxiAAJR5GnVzeg068.jpg

单节点flume配置

flume-1.4.0  启动flume

bin/flume-ng agent --conf ./conf  -f conf/flume-conf.properties -Dflume.root.logger=DEBUG,console -n agent

-n表示配置文件中agent的名字

agent.sources = r1
agent.sinks = s1
agent.channels = c1
agent.sources.r1.channels = c1
agent.sinks.s1.channel = c1

#Describe/configure the sources
agent.sources.r1.type = exec
agent.sources.r1.command = tail -F /home/flume/loginfo
#Use a channel which buffers events in memory
agent.channels.c1.type = memory
agent.channels.c1.capacity = 1000  #Event
agent.channels.c1.transactionCapacity = 100
agent.sinks.s1.type = logger


wKioL1YPs5XgkmmhAAIZ3G0tbb0587.jpg



wKiom1YPs5jyA5Q9AANI-ME5zqU247.jpg


flume-1.4.0 + kafka-0.7.2+hdfs   flume配置


agent.sources = r1
agent.sinks = s_kafka s_hdfs
agent.channels = c_kafka c_hdfs
agent.sources.r1.channels = c_kafka c_hdfs

agent.sources.r1.type = exec
#下面这个脚本tail某个日志
agent.sources.r1.command = tail -F /home/flume/loginfo
agent.channels.c_kafka.type = memory
agent.channels.c_hdfs.type = memory

agent.sinks.s_kafka.type = com.sink.FirstkafkaSink
agent.sinks.s_kafka.channel = c_kafka
#kafka需要连接zk,写入broker数据
agent.sinks.s_kafka.zkconnect = localhost:2181
agent.sinks.s_kafka.topic = test
agent.sinks.s_kafka.serializer.class = kafka.serializer.StringEncoder
agent.sinks.s_kafka.metadata.broker.list = localhost:9092 #配置文件server.properties
agent.sinks.s_kafka.custom.encoding = UTF-8


agent.sinks.s_hdfs.type = hdfs
agent.sinks.s_hdfs.channel = c_hdfs
#默认端口8020
agent.sinks.s_hdfs.hdfs.path = hdfs://localhost:9000/root/source
agent.sinks.s_hdfs.hdfs.filePrefix = events-
agent.sinks.s_hdfs.hdfs.fileType = DataStream
agent.sinks.s_hdfs.hdfs.writeFormat = Text
agent.sinks.s_hdfs.hdfs.rollCount = 30 #达到某一数值记录生成文件
agent.sinks.s_hdfs.hdfs.rollSize = 0
agent.sinks.s_hdfs.hdfs.rollInterval = 0
agent.sinks.s_hdfs.hdfs.useLocalTimeStamp = true
agent.sinks.s_hdfs.hdfs.idleTimeout = 51
agent.sinks.s_hdfs.hdfs.threadsPoolSize = 2


Flume内置channel,source,sink汇总

http://www.iteblog.com/archives/948



本文出自 “点滴积累” 博客,请务必保留此出处http://tianxingzhe.blog.51cto.com/3390077/1700049

【云栖快讯】你想见的Java技术专家都在这了,向大佬提问,有问题必答  详情请点击

网友评论

神巧合
文章365篇 | 关注19
关注
为您提供简单高效、处理能力可弹性伸缩的计算服务,帮助您快速构建更稳定、安全的应用,提升运维效... 查看详情
RDS是一种稳定可靠、可弹性伸缩的在线数据库服务。支持MySQL、SQL Server、Po... 查看详情
阿里云总监课正式启航

阿里云总监课正式启航