spark sql讀寫hive的程序

Spark sql讀寫hive需要hive相關的配置，所以一般將hive-site.xml檔案放到spark的conf目錄下，代碼呼叫都是簡單的，關鍵是原始碼分析程序，spark是如何與hive互動的，

1. 代碼呼叫

讀取hive代碼

SparkSession sparkSession = SparkSession.builder()
                           .appName("read_hive").enableHiveSupport().getOrCreate();
Dataset<Row> data = sparkSession.sql(sqlText); //select 陳述句即可 data就是讀取的表資料集

寫hive代碼

SparkSession sparkSession = SparkSession.builder()
                           .appName("write_hive").enableHiveSupport().getOrCreate();
/*初始化要寫入hive表的資料集
可以是讀取檔案 sparkSession.read().text/csv/parquet()
或者讀取jdbc表sparkSession.read().format("jdbc").option(...).load()
*/
Dataset<Row> data = xxx;
data.createOrReplaceTempView("srcTable"); //創建臨時表
sparkSession.sql("insert into tablex select c1,c2... from srcTable") //將臨時表資料寫入tablex表

注意如果是寫parquet格式的表，要使hivesql也能訪問，則需要在SparkSession上加個配置項 .config("spark.sql.parquet.writeLegacyFormat", true)，這樣hivesql才能訪問，不然會報錯，

2. 原始碼相關的分析

spark sql與hive相關的原始碼就在以下目錄：

對于spark sql的執行流程這里不再介紹，整體架構就是：

讀寫hive的關鍵操作就是enableHiveSupport()方法，在里面會首先檢查是否已經加載了hive的類，然后設定配置項spark.sql.catalogImplementation值為hive，這樣在Sparksession初始化SessionState物件時，根據配置獲取到的就是hive相關的HiveSessionStateBuilder，然后呼叫build創建hive感知的SessionState，

/**
     * Enables Hive support, including connectivity to a persistent Hive metastore, support for
     * Hive serdes, and Hive user-defined functions.
     *
     * @since 2.0.0
     */
    def enableHiveSupport(): Builder = synchronized {
      if (hiveClassesArePresent) {
        config(CATALOG_IMPLEMENTATION.key, "hive")
      } else {
        throw new IllegalArgumentException(
          "Unable to instantiate SparkSession with Hive support because " +
            "Hive classes are not found.")
      }
    }

/**
   * State isolated across sessions, including SQL configurations, temporary tables, registered
   * functions, and everything else that accepts a [[org.apache.spark.sql.internal.SQLConf]].
   * If `parentSessionState` is not null, the `SessionState` will be a copy of the parent.
   *
   * This is internal to Spark and there is no guarantee on interface stability.
   *
   * @since 2.2.0
   */
  @InterfaceStability.Unstable
  @transient
  lazy val sessionState: SessionState = {
    parentSessionState
      .map(_.clone(this))
      .getOrElse {
        val state = SparkSession.instantiateSessionState(
          SparkSession.sessionStateClassName(sparkContext.conf),
          self)
        initialSessionOptions.foreach { case (k, v) => state.conf.setConfString(k, v) }
        state
      }
  }
 
/**
   * Helper method to create an instance of `SessionState` based on `className` from conf.
   * The result is either `SessionState` or a Hive based `SessionState`.
   */
  private def instantiateSessionState(
                                       className: String,
                                       sparkSession: SparkSession): SessionState = {
    try {
      // invoke `new [Hive]SessionStateBuilder(SparkSession, Option[SessionState])`
      val clazz = Utils.classForName(className)
      val ctor = clazz.getConstructors.head
      ctor.newInstance(sparkSession, None).asInstanceOf[BaseSessionStateBuilder].build()
    } catch {
      case NonFatal(e) =>
        throw new IllegalArgumentException(s"Error while instantiating '$className':", e)
    }
  }

private def sessionStateClassName(conf: SparkConf): String = {
    conf.get(CATALOG_IMPLEMENTATION) match {
      case "hive" => HIVE_SESSION_STATE_BUILDER_CLASS_NAME
      case "in-memory" => classOf[SessionStateBuilder].getCanonicalName
    }
  }

SessionState的創建通過BaseSessionStateBuilder.build()來創建

/**
   * Build the [[SessionState]].
   */
  def build(): SessionState = {
    new SessionState(
      session.sharedState,
      conf,
      experimentalMethods,
      functionRegistry,
      udfRegistration,
      () => catalog,
      sqlParser,
      () => analyzer,
      () => optimizer,
      planner,
      streamingQueryManager,
      listenerManager,
      () => resourceLoader,
      createQueryExecution,
      createClone)
  }
}

hive感知的SessionState是通過HiveSessionStateBuilder來創建的，HiveSessionStateBuilder繼承BaseSessionStateBuilder，即相應的catalog/analyzer/planner等都會被HiveSessionStateBuilder重寫的變數或方法代替，
下面將分析HiveSessionCatalog/Analyzer/SparkPlanner

HiveSessionCatalog
SessionCatalog只是一個代理類，只提供呼叫的介面，真正與底層系統互動的是ExternalCatalog，而在hive場景下，HiveSessionCatalog繼承于SessionCatalog，HiveExternalCatalog繼承于ExternalCatalog，

可以看以下類說明：

/**
 * An internal catalog that is used by a Spark Session. This internal catalog serves as a
 * proxy to the underlying metastore (e.g. Hive Metastore) and it also manages temporary
 * views and functions of the Spark Session that it belongs to.
 *
 * This class must be thread-safe.
 */
class SessionCatalog(
    val externalCatalog: ExternalCatalog,
    globalTempViewManager: GlobalTempViewManager,
    functionRegistry: FunctionRegistry,
    conf: SQLConf,
    hadoopConf: Configuration,
    parser: ParserInterface,
    functionResourceLoader: FunctionResourceLoader) extends Logging {
	

/**
 * Interface for the system catalog (of functions, partitions, tables, and databases).
 *
 * This is only used for non-temporary items, and implementations must be thread-safe as they
 * can be accessed in multiple threads. This is an external catalog because it is expected to
 * interact with external systems.
 *
 * Implementations should throw [[NoSuchDatabaseException]] when databases don't exist.
 */
abstract class ExternalCatalog
  extends ListenerBus[ExternalCatalogEventListener, ExternalCatalogEvent] {
  import CatalogTypes.TablePartitionSpec

在HiveExternalCatalog 中，對資料庫、資料表、資料磁區和注冊函式等資訊的讀取與操作都通過 HiveClient 完成， Hive Client 是用來與 Hive 進行互動的客戶端，在 Spark SQL 中是定義了各種基本操作的介面，具體實作為 HiveClientimpl 物件，然而在實際場景中，因為歷史遺留的原因，往往會涉及多種Hive版本，為了有效地支持不同版本，Spark SQL HiveClient的實作由HiveShim通過適配Hive 版本號（HiveVersion）來完成，
在HiveExternalCatalog 中有創建HiveClient的操作，但是最終是呼叫了IsolatedClientLoader來創建，一般spark sql只會通過HiveClient來訪問Hive中的類，為了更好的隔離，IsolatedClientLoader 將不同的類分成3種，不同種類的加載和訪問規則各不相同：
-共享類(Shared classes)：包括基本的Java、Scala Logging和Spark 中的類，這些類通過當前背景關系的 ClassLoader 加載，呼叫 HiveClient 回傳的結果對于外部來說是可見的，
-Hive類(Hive classes)：通過加載 Hive 的相關 Jar 包得到的類，默認情況下，加載這些類的ClassLoader 和加載共享類的 ClassLoader 并不相同，因此，無法在外部訪問這些類
-橋梁類(Barrier classes)：一般包括 HiveClientlmpl和Shim 類，在共享類與 Hive 類之間起到了橋梁的作用，Spark SQL 能夠通過這個類訪問 Hive 中的類，每個新的 HiveClientlmpl實體都對應一個特定的 Hive 版本，

Analyzer
邏輯執行計劃，有著特定于hive的分析規則，
在hive場景中，比基礎的多了ResolveHiveSerdeTable、DetermineTableStats、RelationConversions、HiveAnalysis規則，

SparkPlanner
物理執行計劃，有著特定于hive的策略，
在hive場景中，比基礎的多了HiveTableScans, Scripts策略，

HiveTableScans最終對應的節點HiveTableScanExec，執行hive表的scan操作，磁區屬性和
曬篩選謂詞都可以下推到這里，
Spark sql經過Catalyst的決議，最終轉化成的物理執行計劃，與hive相關的TreeNode主要就是HiveTableScanExec（讀資料）和InsertIntoHiveTable（寫資料），下面主要介紹下這兩個類的實作原理，
HiveTableScanExec
HiveTableScanExec的構造方法引數中比較重要的有兩個，
Relation（HiveTableRelation）, partitionPruningPred（Seq[Expression]）
relation中有著hive表相關的資訊，而partitionPruningPred中有著hive磁區相關的謂詞，
讀取是由hadoopReader（HadoopTableReader）來進行的，不是磁區表則執行
hadoopReader.makeRDDForTable，是磁區表則執行hadoopReader.makeRDDForPartitionedTable，
makeRDDForTable里根據hive表的資料目錄位置創建HadoopRDD，再呼叫
HadoopTableReader.fillObject將原始的Writables資料轉化成Rows，

InsertIntoHiveTable

InsertIntoHiveTable的執行流程就是獲取到HiveExternalCatalog、hadoop相關的配置、hive

表資訊、臨時寫入的目錄位置等，然后呼叫processInsert方法插入，最終再洗掉臨時寫入位

置，processInsert方法里會依次呼叫saveAsHiveFile將RDD寫到臨時目錄檔案中，然后再調

用HiveExternalCatalog的loadTable方法（HiveClient.loadTable -> HiveShim.loadTable -> Hive.loadTable即最侄訓通過反射呼叫Hive的loadTable方法）將臨時寫入目錄位置的檔案

加載到hive表中，

在上面讀寫的程序中，就會涉及到Sparksql Row與Hive資料型別的映射，該轉換功能主要
就是由HiveInspectors來實作，