flume 从log4j 收集日志 到kafka

flume 从log4j 收集日志 到kafka

1. flume 配置

# Define a memory channel called ch1 on agent1agent1.channels.ch1.type = memoryagent1.channels.ch1.capacity = 1000agent1.channels.ch1.transactionCapacity = 100 agent1.sources.avro-source1.channels = ch1agent1.sources.avro-source1.type = avroagent1.sources.avro-source1.bind = localhostagent1.sources.avro-source1.port = 44445 agent1.sinks.kafka-sink1.channel = ch1agent1.sinks.kafka-sink1.type = org.apache.flume.sink.kafka.KafkaSinkagent1.sinks.kafka-sink1.kafka.bootstrap.servers = localhost:9092agent1.sinks.kafka-sink1.topic = testagent1.sinks.kafka-sink1.flumeBatchSize = 10agent1.sinks.kafka-sink1.kafka.producer.acks = 1 # Finally, now that we've defined all of our components, tell# agent1 which ones we want to activate.agent1.channels = ch1agent1.sources = avro-source1agent1.sinks = kafka-sink1

source是avro类型 sink 是kafka的类型

2. 启动flume

flume-ng agent --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/log4g.conf --name agent1 -Dflume.root.logger=INFO,console

3. log4j 打印测试日志

import org.apache.log4j.Logger;public class LoggerGenerator { private static Logger logger = Logger.getLogger(LoggerGenerator.class.getName()); public static void main(String[] args) throws Exception{ int index = 0; while (true){ Thread.sleep(1000); logger.info("value:" + index++); } }}

4. resources log4j 配置

log4j.rootLogger=INFO,stdout,flumelog4j.appender.stdout = org.apache.log4j.ConsoleAppenderlog4j.appender.stdout.target = System.outlog4j.appender.stdout.layout=org.apache.log4j.PatternLayoutlog4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} [%t] [%c] [%p] - %m%nlog4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppenderlog4j.appender.flume.Hostname = localhostlog4j.appender.flume.Port = 44445log4j.appender.flume.UnsafeMode = true

5. marven 依赖

org.apache.flume.flume-ng-clients flume-ng-log4jappender 1.8.0

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。