Flume 读取JMS 消息队列消息,并将消息写入HDFS
2015-08-18 10:00
639 查看
利用Apache Flume 读取JMS 消息队列消息,并将消息写入HDFS,flume agent配置如下:
flume-agent.conf
#name the components on this agent
agentHdfs.sources = jms_source
agentHdfs.sinks = hdfs_sink
agentHdfs.channels = mem_channel
# Describe/configure the source
agentHdfs.sources.jms_source.type = jms
# Bind to all interfaces
agentHdfs.sources.jms_source.initialContextFactory = org.apache.activemq.jndi.ActiveMQInitialContextFactory
agentHdfs.sources.jms_source.connectionFactory = ConnectionFactory
agentHdfs.sources.jms_source.destinationName = BUSINESS_DATA #AMQ queue
agentHdfs.sources.jms_source.providerURL = tcp://hadoop-master:61616
agentHdfs.sources.jms_source.destinationType = QUEUE
# Describe the sink
agentHdfs.sinks.hdfs_sink.type = hdfs
agentHdfs.sinks.hdfs_sink.hdfs.path hdfs://hadoop-master/data/flume/%Y-%m-%d/%H
agentHdfs.sinks.hdfs_sink.hdfs.filePrefix = %{hostname}/events-
agentHdfs.sinks.hdfs_sink.hdfs.maxOpenFiles = 5000
agentHdfs.sinks.hdfs_sink.hdfs.batchSize= 500
agentHdfs.sinks.hdfs_sink.hdfs.fileType = DataStream
agentHdfs.sinks.hdfs_sink.hdfs.writeFormat =Text
agentHdfs.sinks.hdfs_sink.hdfs.rollSize = 0
agentHdfs.sinks.hdfs_sink.hdfs.rollCount = 1000000
agentHdfs.sinks.hdfs_sink.hdfs.rollInterval = 600
agentHdfs.sinks.hdfs_sink.hdfs.useLocalTimeStamp = true
# Use a channel which buffers events in memory
agentHdfs.channels.mem_channel.type = memory
agentHdfs.channels.mem_channel.capacity = 1000
agentHdfs.channels.mem_channel.transactionCapacity = 100
# Bind the source and sink to the channel
agentHdfs.sources.jms_source.channels = mem_channel
agentHdfs.sinks.hdfs_sink.channel = mem_channel
flume-agent.conf
#name the components on this agent
agentHdfs.sources = jms_source
agentHdfs.sinks = hdfs_sink
agentHdfs.channels = mem_channel
# Describe/configure the source
agentHdfs.sources.jms_source.type = jms
# Bind to all interfaces
agentHdfs.sources.jms_source.initialContextFactory = org.apache.activemq.jndi.ActiveMQInitialContextFactory
agentHdfs.sources.jms_source.connectionFactory = ConnectionFactory
agentHdfs.sources.jms_source.destinationName = BUSINESS_DATA #AMQ queue
agentHdfs.sources.jms_source.providerURL = tcp://hadoop-master:61616
agentHdfs.sources.jms_source.destinationType = QUEUE
# Describe the sink
agentHdfs.sinks.hdfs_sink.type = hdfs
agentHdfs.sinks.hdfs_sink.hdfs.path hdfs://hadoop-master/data/flume/%Y-%m-%d/%H
agentHdfs.sinks.hdfs_sink.hdfs.filePrefix = %{hostname}/events-
agentHdfs.sinks.hdfs_sink.hdfs.maxOpenFiles = 5000
agentHdfs.sinks.hdfs_sink.hdfs.batchSize= 500
agentHdfs.sinks.hdfs_sink.hdfs.fileType = DataStream
agentHdfs.sinks.hdfs_sink.hdfs.writeFormat =Text
agentHdfs.sinks.hdfs_sink.hdfs.rollSize = 0
agentHdfs.sinks.hdfs_sink.hdfs.rollCount = 1000000
agentHdfs.sinks.hdfs_sink.hdfs.rollInterval = 600
agentHdfs.sinks.hdfs_sink.hdfs.useLocalTimeStamp = true
# Use a channel which buffers events in memory
agentHdfs.channels.mem_channel.type = memory
agentHdfs.channels.mem_channel.capacity = 1000
agentHdfs.channels.mem_channel.transactionCapacity = 100
# Bind the source and sink to the channel
agentHdfs.sources.jms_source.channels = mem_channel
agentHdfs.sinks.hdfs_sink.channel = mem_channel
相关文章推荐
- HDFS原理
- hadoop点击AM所在node节点的logs链接报错 "Sorry, got error 404"
- HDFS snapshot操作实战
- HDFS优缺点
- HDFS源码分析(二)-----元数据备份机制
- HDFS实现其高可靠性的策略及机制
- [转载] HDFS简介
- hadoop(2.5,2.6) HDFS偶发性心跳异常以及大量DataXceiver线程被Blocked故障处理分享
- 利用mapreduce将数据从hdfs导入到hbase遇到的问题
- 如何在spark中删除hdfs的某个文件夹
- [转载] Kafka+Storm+HDFS整合实践
- HDFS原理及其操作
- MapReduce将HDFS文本数据导入HBase中
- spark implementation hadoop setup,cleanup
- HDFS副本放置策略和机架感知
- flume-ng+Kafka+Storm+HDFS 实时系统搭建
- Hadoop和大数据:60款顶级开源工具
- HDFS文件操作
- HDFS客户端的权限错误:Permission denied
- HDFS客户端