七.Spark Streaming之Flume整合kafka实现实时数据采集

谁借莪１个温暖的怀抱￠ 2022-04-14 03:18 238阅读 0赞

**一.引言**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70]  
    实现架构就是avro sink发送到avro source接收,然后发送到kafka sink作为生产者,另外创建一个消费者进行消费即可,但是一个前提就是机器一监控的是data.log文件的日志内容。

**二.Kafka配置及启动**

host.name=hadoop000
    log.dirs=/root/data/tmp/kafka-logs
    zookeeper.connect=hadoop000:2181

解释上面配置:  
标注主机,日志文件存储路径,注册中心

详细的配置参考[链接][Link 1]

启动kafka:

kafka-server-start.sh $KAFKA_HOME/config/server.properties

**三.Flume配置及启动**

1). 机器一的配置,在Flume的conf目录下创建一个名为exec-memory-avro.conf的配置文件,然后加入以下配置:

exec-memory-avro.sources = exec-source
    exec-memory-avro.sinks = avro-sink
    exec-memory-avro.channels = memory-channel
    
    
    exec-memory-avro.sources.exec-source.type = exec
    exec-memory-avro.sources.exec-source.command = tail -F /root/data/data.log   配置监控文件
    exec-memory-avro.sources.exec-source.shell = /bin/sh -c
    
    exec-memory-avro.sinks.avro-sink.type = avro
    exec-memory-avro.sinks.avro-sink.hostname = 47.100.176.47
    exec-memory-avro.sinks.avro-sink.port = 55555
    
    exec-memory-avro.channels.memory-channel.type = memory
    
    exec-memory-avro.sources.exec-source.channels = memory-channel
    exec-memory-avro.sinks.avro-sink.channel = memory-channel

2).机器二的配置,在Flume目录下创建一个名为avro-memory-kafka.conf的配置文件,然后加入以下配置:

avro-memory-kafka.sources = avro-source
    avro-memory-kafka.sinks = kafka-sink
    avro-memory-kafka.channels = memory-channel
    
    avro-memory-kafka.sources.avro-source.type = avro
    avro-memory-kafka.sources.avro-source.bind = hadoop000
    avro-memory-kafka.sources.avro-source.port = 55555
    
    下面是关于kafkaSink的配置和broker的地址,节点等,
    batchSize是有几条数据控制台更新一次
    avro-memory-kafka.sinks.kafka-sink.type = org.apache.flume.sink.kafka.KafkaSink
    avro-memory-kafka.sinks.kafka-sink.brokerList = hadoop000:9092
    avro-memory-kafka.sinks.kafka-sink.topic = hello_topic
    avro-memory-kafka.sinks.kafka-sink.batchSize = 5
    avro-memory-kafka.sinks.kafka-sink.requiredAcks = 1
    
    avro-memory-kafka.channels.memory-channel.type = memory
    
    avro-memory-kafka.sources.avro-source.channels = memory-channel
    avro-memory-kafka.sinks.kafka-sink.channel = memory-channel

3).两台机器的启动:  
首先要启动机器二,随时准备接收机器一的日志信息:

flume-ng agent --name avro-memory-kafka --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/avro-memory-kafka.conf -Dflume.root.logger=INFO,console

启动成功的标志:  
![在这里插入图片描述][20181123162144409.png]  
然后启动机器一:

flume-ng agent --name exec-memory-avro --conf $FLUME_HOME/conf --conf-file $FLUME_HOME/conf/exec-memory-avro.conf -Dflume.root.logger=INFO,console

启动成功的标志:  
![在这里插入图片描述][20181123162229238.png]

**四.启动消费者和测试**  
1).启动消费者:

kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic hello_topic

2). 测试:  
向Flume监控的data.log文件中写入数据,之后查看kafka是否能消费到:  
![在这里插入图片描述][20181123162604974.png]  
查看消费者是否消费到了(控制台是否打印):  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 1]  
可以看到消费者的控制台已经打印了在data.log中输入的数据。

**五.总结**

上面完成了想data.log中输入数据被Flume收集到,然后发送到kafkaSink,然后kafkaSink作为生产者把收集到的数据供消费者消费,然后启动消费者去消费那个节点,这时就会在控制台上打印出向data.log中输入的数据。Flume和kafka掌握之后可以开始真正的学习Spark Streaming了。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70]: /images/20220414/b8c4a5fb14db4f66a4b1c8649a9eb618.png
[Link 1]: https://blog.csdn.net/ligh_sqh/article/details/84141725
[20181123162144409.png]: /images/20220414/37ff3166da534e9bacc414eb9fca8e07.png
[20181123162229238.png]: /images/20220414/03fea585c6bf4bfbae29ee2cc8ea9296.png
[20181123162604974.png]: /images/20220414/db5d7bf516394862be308db10ca04e15.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 1]: /images/20220414/a7400994cf884a858d09d4ea1f81b3e6.png