流程圖

採集方案

啟動kafka和zk和flume

建立消費者

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

啟動flume

bin/flume-ng agent -n producer -c conf -f conf/myconf.conf -Dflume.root.logger=INFO,console

向flume採集的目標檔案傳送資料:

While true；do echo "我愛你" >> a.log ；sleep 0.5 done；

sparkstreaming消費資料:

package 
 
spark_kafka
import org.apache.spark.streaming._
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

object  Kafka_consumer {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("kafka_wordcount").setMaster("local[*]")
    val ssc = new  StreamingContext(conf,Seconds(2))
    val kafkaParams = Map[String, Object](
    "bootstrap.servers" -> "wangzhihua1:9092,wangzhihua2:9092,wangzhua3:9092",
    "key.deserializer" -> classOf[StringDeserializer],
    "value.deserializer" -> classOf[StringDeserializer],
    "group.id" -> "use_a_separate_group_id_for_each_stream",
    "auto.offset.reset" -> "latest",
    "enable.auto.commit" -> (false: java.lang.Boolean)
  )
    val messages: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](Array("test"), kafkaParams)
    )
    val rs: DStream[(String, Int)] = messages.map(t => {
      (t.value(), 1)
    })
    val fianlRs= rs.reduceByKey(_+_)
    fianlRs.print()
    ssc.start()
    ssc.awaitTermination()
  }
}