今天主要來介紹一下SparkSql,2.x新版本操作hive的一個寫法.

Spark SQL 的功能之一是執行 SQL 查詢.Spark SQL 也能夠被用於從已存在的 Hive 環境中讀取資料

廢話不多說,直接上程式碼:

package spark

import java.io.File
import java.util
import kafka.{PropertiesScalaUtils, RedisKeysListUtils}
import kafka.SparkStreamingKafka.{dbIndex, kafkaStreams}
import net.sf.json.JSONObject
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies}
import redis.RedisPool

/**
  * 利用sparksql 2.0向hive中寫資料;
  */
object SparkSqlDemo {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org.apache.spark").setLevel(Level.INFO)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.INFO)
    Logger.getLogger("org.apache.kafka.clients.consumer").setLevel(Level.INFO)
    val warehouseLocation = new File("hdfs://cluster/hive/warehouse").getAbsolutePath
    val spark = SparkSession.builder().appName("Spark SQL Jason").config("spark.sql.warehouse.dir", warehouseLocation).enableHiveSupport().getOrCreate()
    spark.conf.set("spark.streaming.kafka.maxRatePerPartition", "2000")
    spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    spark.conf.set("spark.streaming.concurrentJobs", "10")
    spark.conf.set("spark.streaming.kafka.maxRetries", "50")
    @transient
    val sc = spark.sparkContext
    val scc = new StreamingContext(sc, Seconds(2))
    val topic = "jason_20180511"
    val topicSet: Set[String] = Set(topic) //設定kafka的topic;
    val kafkaParams = Map[String, Object](
      "auto.offset.reset" -> "latest",
      "value.deserializer" -> classOf[StringDeserializer]
      , "key.deserializer" -> classOf[StringDeserializer]
      , "bootstrap.servers" -> PropertiesScalaUtils.loadProperties("broker")
      , "group.id" -> PropertiesScalaUtils.loadProperties("groupId")
      , "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    val maxTotal = 200
    val maxIdle = 100
    val minIdle = 10
    val testOnBorrow = false
    val testOnReturn = false
    val maxWaitMillis = 500
    RedisPool.makePool(PropertiesScalaUtils.loadProperties("redisHost"), PropertiesScalaUtils.loadProperties("redisPort").toInt, maxTotal, maxIdle, minIdle, testOnBorrow, testOnReturn, maxWaitMillis)
    val jedis = RedisPool.getPool.getResource
    jedis.select(dbIndex)
    val keys: util.Set[String] = jedis.keys(topic + "*")
    if (keys.size() == 0) {
      kafkaStreams = KafkaUtils.createDirectStream[String, String](
        scc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topicSet, kafkaParams))
    } else {
      val fromOffsets: Map[TopicPartition, Long] = RedisKeysListUtils.getKeysList(PropertiesScalaUtils.loadProperties("redisHost"), PropertiesScalaUtils.loadProperties("redisPort").toInt, topic)
      kafkaStreams = KafkaUtils.createDirectStream[String, String](
        scc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topicSet, kafkaParams, fromOffsets))
    }
    RedisPool.getPool.returnResource(jedis)
    kafkaStreams.foreachRDD(rdd=>{
      val jedis_jason = RedisPool.getPool.getResource
      jedis_jason.select(dbIndex)
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      import spark.sql
      if(!rdd.isEmpty()){
        val rowRDD:RDD[Row] = rdd.map(x=>{
          val json = JSONObject.fromObject(x.value().toString)
          val a = json.get("name")
          val b = json.get("addr")
          Row(a,b)
        })
        val schemaString = "name addr"
        val field = schemaString.split(" ").map(x=> StructField(x,StringType,nullable = true))
        val schema = StructType(field)
        val df = spark.createDataFrame(rowRDD, schema)
        df.show()
        df.createOrReplaceTempView("tempTable")
        val sq = "insert into test_2 select * from tempTable"
        sql(sq)
        println("插入hive成功了")
      }
      offsetRanges.foreach { offsetRange =>
        println("partition : " + offsetRange.partition + " fromOffset:  " + offsetRange.fromOffset + " untilOffset: " + offsetRange.untilOffset)
        val topic_partition_key_new = offsetRange.topic + "_" + offsetRange.partition
        jedis_jason.set(topic_partition_key_new, offsetRange.untilOffset + "")
      }
    })
    scc.start()
    scc.awaitTermination()
  }
}
 

需要注意的是: spark.sql.warehouse.dir 配置的目錄，該目錄預設為Spark應用程式當前目錄中的 spark-warehouse 目錄 但請注意,自從2.0.0以來，hive-site.xml 中的 hive.metastore.warehouse.dir 屬性已被棄用。 而是使用 spark.sql.warehouse.dir 來指定倉庫中資料庫的預設位置。

還有一個問題是,這樣寫的話,會在hdfs上這個表的目錄下生成很多的小檔案,這個時候如果想在hive中進行統計,計算的時候,會產生很多個map,嚴重影響計算的速度,大家可以先考慮下這個問題.

為了解決在表目錄下面生成很多小檔案的問題,我們可以把hive表建成一個分割槽表,怎麼建分割槽表在我的另一篇blog裡面有寫到,或者可以直接用: insert overwrite table combine_data partition (day_time='2018-08-01') select data,enter_time from combine_data where day_time = '2018-08-01';來合併小檔案.

或者也可把用reparation減少分割槽數,但是這麼寫,會減少rdd的並行度,降低效能,自己參考使用.

如果有寫的不對的地方,歡迎大家指正,如果有什麼疑問,可以加QQ群:340297350,謝謝

參考:

http://spark.apachecn.org/docs/cn/2.2.0/sql-programming-guide.html點選開啟連結