1、在flume目录中创建flume-kafka-tail-conf.properties文件

# The configuration file needs to define the sources,  # the channels and the sinks. # Sources, channels and sinks are defined per agent,  # in this case called 'agent' a2.sources = r2 a2.channels = c2
a2.sinks = k2 ### define sources a2.sources.r2.type = exec
a2.sources.r2.command = tail -F /opt/datas/spark_word_count.log a2.sources.r2.shell = /bin/bash -c ### define channels a2.channels.c2.type = memory
a2.channels.c2.capacity = 1000 a2.channels.c2.transactionCapacity = 100 ### define sinks a2.sinks.k2.type = org.apache.flume.sink.kafka.KafkaSink a2.sinks.k2.brokerList = bigdata.eclipse.com:9092 a2.sinks.k2.topic = kafka_topic ### bind the sources and sinks to the channels a2.sources.r2.channels=c2
a2.sinks.k2.channel = c2 

	
	

		1
	

	

		2
	

	

		3
	

	

		4
	

	

		5
	

	

		6
	

	

		7
	

	

		8
	

	

		9
	

	

		10
	

	

		11
	

	

		12
	

	

		13
	

	

		14
	

	

		15
	

	

		16
	

	

		17
	

	

		18
	

	

		19
	

	

		20
	

	

		21
	

	

		22
	

	

		23
	

	

		24
	

	

		25
	

	

		26
	

	

		27
	

	

		28
	




	
	

		1
	

	

		2
	

	

		3
	

	

		4
	

	

		5
	

	

		6
	

	

		7
	

	

		8
	

	

		9
	

	

		10
	

	

		11
	

	

		12
	

	

		13
	

	

		14
	

	

		15
	

	

		16
	

	

		17
	

	

		18
	

	

		19
	

	

		20
	

	

		21
	

	

		22
	

	

		23
	

	

		24
	

	

		25
	

	

		26
	

	

		27
	

	

		28
	

2、测试

bin/kafka-topics.sh --create --zookeeper bigdata.eclipse.com:2181 --replication-factor 1 --partitions 1 --topic kafka_topic bin/kafka-topics.sh --list --zookeeper bigdata.eclipse.com:2181 bin/kafka-console-producer.sh --broker-list bigdata.eclipse.com:9092 --topic kafka_topic bin/kafka-console-consumer.sh --zookeeper bigdata.eclipse.com:2181 --topic kafka_topic --from-beginning