🎃生產者&消費者小案例

---

主要通過Flume進行數據的收集，作為數據的生產者，接入kafka作為消息中間件，用SparkStreaming作為消費者進行數據的消費【前提Linux已準備好Flume+Kafka環境，網上很多教程這里不多贅述，直接進入主要內容】

---

1、準備Flume的配置文件

apache-flume-1.9.0-bin/options/exec2kafka.conf文件：

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

---

2、編寫SparkStreaming消費者

object KafkaSpark {def main(args: Array[String]): Unit = {//創建val sparkConf = new SparkConf().setMaster("local[2]").setAppName("Hello07Kafka")val streamingContext = new StreamingContext(sparkConf, Seconds(1))//配置信息val kafkaParams = 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 = Array("baidu")//開始創建Kafkaval linesDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils createDirectStream(streamingContext, PreferConsistent, Subscribe[String, String](topics, kafkaParams))//打印數據linesDStream.map(_.value()).foreachRDD(rdd => 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=INFO,console

最后啟動消費者SparkStreaming代碼~

---

---

4、進行數據測試

① 通過ping baidu進行測試

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

---

② 通過文件追加

[root@node01 ~]# echo helloworld>>/root/baidu.log [root@node01 ~]# echo helloworld1>>/root/baidu.log [root@node01 ~]# echo helloworld2>>/root/baidu.log

---

總結

以上是生活随笔為你收集整理的Flume+Kafka+Spark小案例的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。