Flume+Kafka+Spark小案例

&＃x1f383;生产者&消费者小案例

主要通过Flume进行数据的收集&＃xff0c;作为数据的生产者&＃xff0c;接入kafka作为消息中间件&＃xff0c;用SparkStreaming作为消费者进行数据的消费【前提Linux已准备好Flume&＃43;Kafka环境&＃xff0c;网上很多教程这里不多赘述&＃xff0c;直接进入主要内容】

1、准备Flume的配置文件

apache-flume-1.9.0-bin/options/exec2kafka.conf文件&＃xff1a;

## flume-ng agent -n a1 -c options/ -f exec2kafka.conf -Dflume.root.logger&＃61;INFO,console##定义a1的三个组件的名称
a1.sources &＃61; r1
a1.sinks &＃61; k1
a1.channels &＃61; c1##定义Source的类型
a1.sources.r1.type &＃61; exec
a1.sources.r1.command &＃61; tail -F /root/baidu.log##定义Channel的类型
a1.channels.c1.type &＃61; memory
a1.channels.c1.capacity &＃61; 1000
a1.channels.c1.transactionCapacity &＃61; 100##定义Sink的类型
a1.sinks.k1.type &＃61; org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic &＃61; baidu
a1.sinks.k1.kafka.bootstrap.servers &＃61; 192.168.88.101:9092,192.168.88.102:9092,192.168.88.103:9092
a1.sinks.k1.kafka.flumeBatchSize &＃61; 10
a1.sinks.k1.kafka.producer.acks &＃61; 1
a1.sinks.k1.kafka.producer.linger.ms &＃61; 1##组装source channel sink
a1.sources.r1.channels &＃61; c1
a1.sinks.k1.channel &＃61; c1


2、编写SparkStreaming消费者

object KafkaSpark {def main(args: Array[String]): Unit &＃61; {//创建val sparkConf &＃61; new SparkConf().setMaster("local[2]").setAppName("Hello07Kafka")val streamingContext &＃61; new StreamingContext(sparkConf, Seconds(1))//配置信息val kafkaParams &＃61; Map[String, Object]("bootstrap.servers" -> "node01:9092,node02:9092,node03:9092","key.deserializer" -> classOf[StringDeserializer],"value.deserializer" -> classOf[StringDeserializer],"group.id" -> "yjx_kafka","auto.offset.reset" -> "earliest","enable.auto.commit" -> (false: java.lang.Boolean))//主题val topics &＃61; Array("baidu")//开始创建Kafkaval linesDStream: InputDStream[ConsumerRecord[String, String]] &＃61; KafkaUtils createDirectStream(streamingContext, PreferConsistent, Subscribe[String, String](topics, kafkaParams))//打印数据linesDStream.map(_.value()).foreachRDD(rdd &＃61;> rdd.foreach(println))streamingContext.start()streamingContext.awaitTermination()}
}

3、启动环境

① 启动zookeeper集群

【1】【2】【3】zkServer.sh start


② 启动kafka集群

#启动kafka集群
【1】【2】【3】kafka-server-start.sh /opt/yjx/kafka_2.12-0.11.0.3/config/server.properties
#创建主题
【1】kafka-topics.sh --zookeeper node01:2181 --create --replication-factor 2 --partitions 3 --topic baidu


③ 启动Flume

#flume启动命令
【1】flume-ng agent -n a1 -f /opt/yjx/apache-flume-1.9.0-bin/options/exec2kafka.conf -Dflume.root.logger&＃61;INFO,console


最后启动消费者SparkStreaming代码~

4、进行数据测试

① 通过ping baidu进行测试

#ping结果到baidu.log文件中
ping www.baidu.com >> /root/baidu.log 2>&1 &


② 通过文件追加

[root&＃64;node01 ~]# echo helloworld>>/root/baidu.log
[root&＃64;node01 ~]# echo helloworld1>>/root/baidu.log
[root&＃64;node01 ~]# echo helloworld2>>/root/baidu.log