非kerberos环境下，flume采集日志到kafka

kafkaflume.conf

agent.sources = s1agent.channels = c1agent.sinks = k1agent.sources.s1.type=execagent.sources.s1.command=tail -F /usr/local/src/flume/testflume2.logagent.sources.s1.channels=c1agent.channels.c1.type=memoryagent.channels.c1.capacity=10000agent.channels.c1.transactionCapacity=100#设置Kafka接收器#agent.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink##设置Kafka的broker地址和端口号agent.sinks.k1.kafka.bootstrap.servers = master:9092,node01:9092,node02:9092##设置Kafka的Topicagent.sinks.k1.kafka.topic = mydemo5##设置序列化方式##agent.sinks.k1.kafka.flumeBatchSize = 20agent.sinks.k1.kafka.producer.acks = 1agent.sinks.k1.kafka.producer.linger.ms = 1agent.sinks.ki.kafka.producer.compression.type = snappyagent.sinks.k1.channel=c1

创建一个topic

kafka-topics --create --zookeeper master:2181/kafka --replication-factor 2 --partitions 3 --topic mydemo5

启动flume作为消息生产者写道kafka上

flume-ng agent -c conf -f kafkaflume.conf -n agent -Dflume.root.logger=INFO,console

消费kafka的数据

kafka-console-consumer --bootstrap-server master:9092 --topic mydemo5 --from-beginning

读取一个文件的内容写入追加到新的文件

#!/bin/bashwhile read linedo    echo $line    sleep 0.01    echo -e $line >> /usr/local/src/flume/testflume2.logdone < /etc/sudo-ldap.conf