Kafka+SparkStream+Hive的項(xiàng)目實(shí)現(xiàn)方法是什么

本篇內(nèi)容主要講解“Kafka+SparkStream+Hive的項(xiàng)目實(shí)現(xiàn)方法是什么”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實(shí)用性強(qiáng)。下面就讓小編來帶大家學(xué)習(xí)“Kafka+SparkStream+Hive的項(xiàng)目實(shí)現(xiàn)方法是什么”吧!

目前的項(xiàng)目中需要將kafka隊(duì)列的數(shù)據(jù)實(shí)時(shí)存到hive表中。

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
import org.apache.spark.streaming.{Durations, Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange}
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

  def main(args: Array[String]): Unit = {
      //    val conf = new SparkConf()
      //    conf.setMaster("local")
      //    conf.setAppName("SparkStreamingOnKafkaDirect")
      val spark = SparkSession.builder().appName("test").master("local").enableHiveSupport().getOrCreate()
      val ssc = new StreamingContext(spark.sparkContext, Durations.seconds(3))
      //設(shè)置日志級別
      ssc.sparkContext.setLogLevel("Error")

      val kafkaParams = Map[String, Object](
        "bootstrap.servers" -> "node01:9092,node02:9092,node03:9092",
        "key.deserializer" -> classOf[StringDeserializer],
        "value.deserializer" -> classOf[StringDeserializer],
        "group.id" -> "MyGroupId", //

        /**
         * 當(dāng)沒有初始的offset，或者當(dāng)前的offset不存在，如何處理數(shù)據(jù)
         * earliest ：自動(dòng)重置偏移量為最小偏移量
         * latest：自動(dòng)重置偏移量為最大偏移量【默認(rèn)】
         * none:沒有找到以前的offset,拋出異常
         */
        "auto.offset.reset" -> "earliest",

        /**
         * 當(dāng)設(shè)置 enable.auto.commit為false時(shí)，不會(huì)自動(dòng)向kafka中保存消費(fèi)者offset.需要異步的處理完數(shù)據(jù)之后手動(dòng)提交
         */
        "enable.auto.commit" -> (false: java.lang.Boolean) //默認(rèn)是true
      )

      //設(shè)置Kafka的topic
      val topics = Array("test")
      //創(chuàng)建與Kafka的連接，接收數(shù)據(jù)
      /*這里接收到數(shù)據(jù)的樣子
      2019-09-26  1569487411604   1235    497 Kafka   Register
      2019-09-26  1569487411604   1235    497 Kafka   Register
      2019-09-26  1569487414838   390    778  Flink   View
      */
      val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
        ssc,
        PreferConsistent, //
        Subscribe[String, String](topics, kafkaParams)
      )

      //對接收到的數(shù)據(jù)進(jìn)行處理，打印出來接收到的key跟value，最后放回的是value
      val transStrem: DStream[String] = stream.map(record => {
        val key_value = (record.key, record.value)
        println("receive message key = " + key_value._1)
        println("receive message value = " + key_value._2)
        key_value._2
      })


      //這里用了一下動(dòng)態(tài)創(chuàng)建的Schema
      val structType: StructType = StructType(List[StructField](
        StructField("Date_", StringType, nullable = true),
        StructField("Timestamp_", StringType, nullable = true),
        StructField("UserID", StringType, nullable = true),
        StructField("PageID", StringType, nullable = true),
        StructField("Channel", StringType, nullable = true),
        StructField("Action", StringType, nullable = true)
      ))

      //因?yàn)閒oreachRDD可以拿到封裝到DStream中的rdd，可以對里面的rdd進(jìn)行，
      /*代碼解釋：
          先從foreach中拿到一條數(shù)據(jù)，，在函數(shù)map中對接收來的數(shù)據(jù)用 “\n” 進(jìn)行切分，放到Row中，用的是動(dòng)態(tài)創(chuàng)建Schema，因?yàn)槲覀冃枰賹?shù)據(jù)存儲(chǔ)到hive中，所以需要Schema。
          因?yàn)閙ap是transformance算子，所以用rdd.count（）觸發(fā)一下
           spark.createDataFrame：創(chuàng)建一個(gè)DataFrame，因?yàn)橐砸粋€(gè)臨時(shí)表，必須用到DataFrame
           frame.createOrReplaceTempView("t1")：注冊臨時(shí)表
             spark.sql("use spark")：使用 hive 的 spark 庫
           result.write.mode(SaveMode.Append).saveAsTable("test_kafka")：將數(shù)據(jù)放到 test_kafka 中
      */
      transStrem.foreachRDD(one => {
        val rdd: RDD[Row] = one.map({
          a =>
            val arr = a.toString.split("\t")
            Row(arr(0).toString, arr(1).toString, arr(2).toString, arr(3).toString, arr(4).toString, arr(5).toString)
        })
        rdd.count()
        val frame: DataFrame = spark.createDataFrame(rdd, structType)
        //      println(" Scheme: "+frame.printSchema())

        frame.createOrReplaceTempView("t1")
        //      spark.sql("select * from t1").show()
        spark.sql("use spark")
        spark.sql("select * from t1").
          write.mode(SaveMode.Append).saveAsTable("test_kafka")
      }
      )

      /**
       * 以上業(yè)務(wù)處理完成之后，異步的提交消費(fèi)者offset,這里將 enable.auto.commit 設(shè)置成false,就是使用kafka 自己來管理消費(fèi)者offset
       * 注意這里，獲取 offsetRanges: Array[OffsetRange] 每一批次topic 中的offset時(shí)，必須從 源頭讀取過來的 stream中獲取，不能從經(jīng)過stream轉(zhuǎn)換之后的DStream中獲取。
       */
      stream.foreachRDD { rdd =>
        val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        // some time later, after outputs have completed
        stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
      }
      ssc.start()
      ssc.awaitTermination()
      ssc.stop()
  }

到此，相信大家對“Kafka+SparkStream+Hive的項(xiàng)目實(shí)現(xiàn)方法是什么”有了更深的了解，不妨來實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！