flume段定义
    flume_kafka.conf

    1. a1.sources = r1
    2. a1.sinks = k1
    3. a1.channels = c1
    5. #定义source信息
    6. a1.sources.r1.type=netcat
    7. a1.sources.r1.bind=localhost
    8. a1.sources.r1.port=8888
    10. a1.channels.c1.type=memory
    12. a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
    13. a1.sinks.k1.kafka.topic = test
    14. a1.sinks.k1.brokerList = s2:9092
    15. a1.sinks.k1.kafka.flumeBatchSize = 20
    16. a1.sinks.k1.kafka.producer.acks = 1
    18. a1.sources.r1.channels = c1
    19. a1.sinks.k1.channel = c1

    启动flume

    1. bin/flume-ng agent -f ../conf/flume_kafka.conf -n a1 -Dflume.root.logger=INFO,console

    kafka端数据采集

    生产者-消费者测试

    1. 生产者
    2. kafka-console-producer.sh --broker-list s2:9092 --topic test
    4. 消费者
    5. kafka-console-consumer.sh --bootstrap-server s2:9092 --topic test

    测试方式：

    1. 1.启动nc
    2. [root@s3 ~]# nc localhost 8888
    5. 2.启动flume
    6. [root@s3 flumeconfig]# flume-ng agent -f flume_kafka.conf -n a1 -Dflume.root.logger=INFO,console
    8. 3.启动kafka消费者
    9. [root@s2 ~]# kafka-console-consumer.sh --bootstrap-server s2:9092 --topic test
    11. 在nc输入abc,在kafka消费者有消息显示