1、创建一个agent，sink类型需指定为自定义sink
vi /usr/local/flume/conf/agent3.conf
agent3.sources=as1
agent3.channels=c1
agent3.sinks=s1
agent3.sources.as1.type=avro
agent3.sources.as1.bind=0.0.0.0
agent3.sources.as1.port=41414
agent3.sources.as1.channels=c1
agent3.channels.c1.type=memory
agent3.sinks.s1.type=storm.test.kafka.TestKafkaSink
agent3.sinks.s1.channel=c1
2、创建自定义kafka sink（自定义kafka sink中包装的是kafka的生产者），代码如下
//参考flume官方的开发文档：http://flume.apache.org/FlumeDeveloperGuide.html#sink
//自定义kafkasink需要继承AbstractSink类实现Configurable接口
//该sink中使用的kafka topic（test111）必须存在
package storm.test.kafka;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
public class TestKafkaSink extends AbstractSink implements Configurable {
Producer<String, String> producer;
String topic = "test111";
@Override
public Status process() throws EventDeliveryException {
Status status = null;
Channel channel = getChannel();
Transaction transaction = channel.getTransaction();
transaction.begin();
try {
Event event = channel.take();
if (event==null) {
transaction.rollback();
status = Status.BACKOFF;
return status;
}
byte[] body = event.getBody();
final String msg = new String(body);
final KeyedMessage<String, String> message = new KeyedMessage<String, String>(topic , msg);
producer.send(message);
transaction.commit();
status = Status.READY;
} catch (Exception e) {
transaction.rollback();
status = Status.BACKOFF;
} finally {
transaction.close();
}
return status;
}
@Override
public void configure(Context arg0) {
Properties prop = new Properties();
prop.put("zookeeper.connect", "h5:2181,h6:2181,h7:2181");
prop.put("metadata.broker.list", "h5:9092,h6:9092,h7:9092");
prop.put("serializer.class", StringEncoder.class.getName());
producer = new Producer<String, String>(new ProducerConfig(prop));
}
}
将代码打包为kafkasink.jar后复制到flume所在节点上的flume/lib目录下，然后还需要将kafka_2.10-0.8.2.0.jar、kafka-clients-0.8.2.0.jar、metrics-core-2.2.0.jar、scala-library-2.10.4.jar这4个jar包复制到flume所在节点上的flume/lib目录下。
3、启动flume自定义的kafkasink的agent
[root@h5 ~]# cd /usr/local/flume/
[root@h5 flume]# bin/flume-ng agent --conf conf/ --conf-file conf/agent3.conf --name agent3 -Dflume.root.logger=INFO,console
4、将日志写入到flume的agent，代码如下
log4j.properties
log4j.rootLogger=INFO,flume
log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender
log4j.appender.flume.Hostname = 192.168.1.35
log4j.appender.flume.Port = 41414
log4j.appender.flume.UnsafeMode = true
将日志写入到flume，代码如下
package com.mengyao.flume;
import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import org.apache.commons.io.FileUtils;
import org.apache.log4j.Logger;
public class FlumeProducer {
private static List<String> getLines() {
List<String> lines = null;
try {
final Collection<File> listFiles = FileUtils.listFiles(new File("D:/"), null, false);
for (File file : listFiles) {
lines = FileUtils.readLines(file);
break;
}
} catch (IOException e) {
e.printStackTrace();
}
return lines;
}
public static void main(String[] args) throws Exception {
final List<String> lines = getLines();
final Logger logger = Logger.getLogger(FlumeProducer.class);
for (String line : lines) {
logger.info(line+"\t"+System.currentTimeMillis());
Thread.sleep(1000);
}
}
}
必须加入flume-ng-log4jappender-1.5.0-cdh5.1.3-jar-with-dependencies.jar这个依赖jar
5、使用kafka消费者消费flume（自定义kafka sink中使用了kafka的生产者）生产的数据
1、消费者shell代码
[root@h7 kafka]# bin/kafka-console-consumer.sh --zookeeper h7:2181 --topic test111 --from-beginning ##kafka集群是h5、h6、h7;zookeeper集群是h5、h6、h7。在任意kafka节点上使用消费者都一样
2、消费者java代码
package storm.test.kafka;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringEncoder;
public class TestConsumer {
static final String topic = "test111";
public static void main(String[] args) {
Properties prop = new Properties();
prop.put("zookeeper.connect", "h5:2181,h6:2181,h7:2181");
prop.put("serializer.class", StringEncoder.class.getName());
prop.put("metadata.broker.list", "h5:9092,h6:9092,h7:9092");
prop.put("group.id", "group1");
ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(prop));
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(topic, 1);
Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
final KafkaStream<byte[], byte[]> kafkaStream = messageStreams.get(topic).get(0);
ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
while (iterator.hasNext()) {
String msg = new String(iterator.next().message());
System.out.println("收到消息："+msg);
}
}
}