SprakSQL-Catalog

祝福

在这个举国同庆的时刻，我们首先献上对祖国的祝福：

第一，我们感谢您给我们和平的环境，让我们能快乐生活

第二，祝福我们国家未来的路越走越宽广，科技更发达，人民更幸福

第三，我们会紧紧跟随您的脚步，一起为美好的未来奋斗

一、概述

Catalog可以翻译为目录，意思就是用户在使用的时候可以通过它大致了解整个数据的框架和结构，比如：充当底层元存储（例如Hive元存储）的代理、管理其所属Spark会话的临时视图和函数。

二、使用

val spark = SparkSession
  .builder()
  .appName("Spark Hive Example")
  .config("spark.sql.warehouse.dir", warehouseLocation)
  .enableHiveSupport()
  .getOrCreate()

import spark.implicits._
import spark.sql

//列举表
spark.catalog.listTables().show()
//缓存表
spark.catalog.cacheTable("tableName")
//释放表
spark.catalog.uncacheTable("tableName")

三、源码

1、SparkSession

构建一个CatalogImpl，用户可以通过该界面创建、删除、更改或查询底层数据库、表、函数等。

@transient lazy val catalog: Catalog = new CatalogImpl(self)

2、CatalogImpl

CatalogImpl是面向用户的“目录”的内部实现

class CatalogImpl(sparkSession: SparkSession) extends Catalog {

  private def sessionCatalog: SessionCatalog = sparkSession.sessionState.catalog

  //返回此会话中的当前默认数据库
  override def currentDatabase: String = sessionCatalog.getCurrentDatabase

  //返回所有会话中可用的数据库列表
  override def listDatabases(): Dataset[Database] = {
    val databases = sessionCatalog.listDatabases().map(makeDatabase)
    CatalogImpl.makeDataset(databases, sparkSession)
  }

  //创建一个新的数据库
  private def makeDatabase(dbName: String): Database = {
    val metadata = sessionCatalog.getDatabaseMetadata(dbName)
    new Database(
      name = metadata.name,
      description = metadata.description,
      locationUri = CatalogUtils.URIToString(metadata.locationUri))
  }

  //返回当前数据库中的表列表。这包括所有临时表。
  override def listTables(): Dataset[Table] = {
    listTables(currentDatabase)
  }

  //返回指定数据库中的表列表。这包括所有临时表。
  @throws[AnalysisException]("database does not exist")
  override def listTables(dbName: String): Dataset[Table] = {
    val tables = sessionCatalog.listTables(dbName).map(makeTable)
    CatalogImpl.makeDataset(tables, sparkSession)
  }


  //返回给定表/视图或临时视图的表。
  //请注意，此函数要求该表已存在于目录中。
  //如果由于任何原因导致表元数据检索失败（例如，表serde类不可访问或Spark SQL不接受表类型），此函数仍将返回相应的表，而不包含描述和表类型）
  private def makeTable(tableIdent: TableIdentifier): Table = {
    val metadata = try {
      Some(sessionCatalog.getTempViewOrPermanentTableMetadata(tableIdent))
    } catch {
      case NonFatal(_) => None
    }
    val isTemp = sessionCatalog.isTempView(tableIdent)
    new Table(
      name = tableIdent.table,
      database = metadata.map(_.identifier.database).getOrElse(tableIdent.database).orNull,
      description = metadata.map(_.comment.orNull).orNull,
      tableType = if (isTemp) "TEMPORARY" else metadata.map(_.tableType.name).orNull,
      isTemporary = isTemp)
  }

  //返回当前数据库中注册的函数列表。这包括所有临时功能
  override def listFunctions(): Dataset[Function] = {
    listFunctions(currentDatabase)
  }

  //注册一个自定义的函数
  private def makeFunction(funcIdent: FunctionIdentifier): Function = {
    val metadata = sessionCatalog.lookupFunctionInfo(funcIdent)
    new Function(
      name = metadata.getName,
      database = metadata.getDb,
      description = null, // for now, this is always undefined
      className = metadata.getClassName,
      isTemporary = metadata.getDb == null)
  }

  //返回给定表/视图或临时视图的列列表。
  @throws[AnalysisException]("table does not exist")
  override def listColumns(tableName: String): Dataset[Column] = {
    val tableIdent = sparkSession.sessionState.sqlParser.parseTableIdentifier(tableName)
    listColumns(tableIdent)
  }

  //........


}

从CatalogImpl中我们可以看到有这样的功能：查看数据库、表、列、视图、函数列表、创建数据库、表、视图、函数、缓存表、释放表等等，其中基本都使用了SessionCatalog，下面我们详细看下它。

3、SessionCatalog

object SessionCatalog {
  val DEFAULT_DATABASE = "default"
}
class SessionCatalog(
    externalCatalogBuilder: () => ExternalCatalog,
    globalTempViewManagerBuilder: () => GlobalTempViewManager,
    functionRegistry: FunctionRegistry,
    tableFunctionRegistry: TableFunctionRegistry,
    hadoopConf: Configuration,
    parser: ParserInterface,
    functionResourceLoader: FunctionResourceLoader,
    cacheSize: Int = SQLConf.get.tableRelationCacheSize,
    cacheTTL: Long = SQLConf.get.metadataCacheTTL) extends SQLConfHelper with Logging {

  lazy val externalCatalog = externalCatalogBuilder()
  lazy val globalTempViewManager = globalTempViewManagerBuilder()

  //临时视图列表，从表名映射到其逻辑计划
  @GuardedBy("this")
  protected val tempViews = new mutable.HashMap[String, TemporaryViewRelation]

  //注意：我们在这里跟踪当前数据库，因为某些操作没有明确指定数据库（例如DROP TABLE my_TABLE）。在这些情况下，我们必须首先检查临时视图或函数是否存在，如果不存在，则对当前数据库中的相应项进行操作。
  @GuardedBy("this")
  protected var currentDb: String = formatDatabaseName(DEFAULT_DATABASE)

  private val validNameFormat = "([\\w_]+)".r

  //检查给定名称是否符合Hive标准（“[a-zA-Z_0-9]+”），即此名称是否仅包含字符、数字和_。
  //此方法旨在具有与org.apache.hoop.hive.metastore相同的行为。MetaStoreUtils.validateName。
  private def validateName(name: String): Unit = {
    if (!validNameFormat.pattern.matcher(name).matches()) {
      throw QueryCompilationErrors.invalidNameForTableOrDatabaseError(name)
    }
  }

  //设置表名格式，同时考虑区分大小写。
  protected[this] def formatTableName(name: String): String = {
    if (conf.caseSensitiveAnalysis) name else name.toLowerCase(Locale.ROOT)
  }

  protected[this] def formatDatabaseName(name: String): String = {
    if (conf.caseSensitiveAnalysis) name else name.toLowerCase(Locale.ROOT)
  }

  //获取缓存计划
  def getCachedPlan(t: QualifiedTableName, c: Callable[LogicalPlan]): LogicalPlan = {
    tableRelationCache.get(t, c)
  }

  //下面是对数据库、表、分区、函数的操作
  //............

}

我们来整体看下它有哪些功能：

        1、数据库：此类别中的所有方法都直接与底层Catalog交互

        2、表：有两种表，临时视图和元存储表。

                     临时视图在会话之间是隔离的，不属于任何特定的数据库。

                     元存储表可以在多个会话中使用，因为它们的元数据被持久化在底层Catalog中。

                     与元存储表交互的方法：

                        createTable() 、alterTable() 、alterTableDataSchema() 、tableExists() 等

                     与临时视图交互的方法：

                        createTempView() 、createGlobalTempView() 、 alterTempViewDefinition()等

        3、分区：加载数据源表时，将自动发现表的分区

        4、函数：临时函数和元存储函数（永久UDF）

                        临时功能在会话之间隔离。元存储函数可以在多个会话中使用，

                        因为它们的元数据保存在底层Catalog中。

数据库

我们以创建数据库为例来看下内部逻辑

def createDatabase(dbDefinition: CatalogDatabase, ignoreIfExists: Boolean): Unit = {
    //设置数据库名称的格式，同时考虑区分大小写
    val dbName = formatDatabaseName(dbDefinition.name)
    //校验该数据库名释放占用
    if (dbName == globalTempViewManager.database) {
      throw QueryCompilationErrors.cannotCreateDatabaseWithSameNameAsPreservedDatabaseError(
        globalTempViewManager.database)
    }
    //校验数据库名释放符合命名规范
    validateName(dbName)
    //调用外部的Catalog创建数据库
    externalCatalog.createDatabase(
      dbDefinition.copy(name = dbName, locationUri = makeQualifiedDBPath(dbDefinition.locationUri)),
      ignoreIfExists)
  }

表

元存储表

我们以创建表为例来看下内部逻辑

  //在“tableDefinition”中指定的数据库中创建一个元存储表。如果没有指定此类数据库，请在当前数据库中创建它。
  def createTable(
      tableDefinition: CatalogTable,
      ignoreIfExists: Boolean,
      validateLocation: Boolean = true): Unit = {
    val isExternal = tableDefinition.tableType == CatalogTableType.EXTERNAL
    if (isExternal && tableDefinition.storage.locationUri.isEmpty) {
      throw QueryCompilationErrors.createExternalTableWithoutLocationError
    }
    
    //获取数据库名
    val db = formatDatabaseName(tableDefinition.identifier.database.getOrElse(getCurrentDatabase))
    //格式化表名 ，默认不打开大小写区分，但官方强烈建议打开，且设置成小写去解析
    val table = formatTableName(tableDefinition.identifier.table)
    //在数据库中标识这个表
    val tableIdentifier = TableIdentifier(table, Some(db))
    //校验表名释放符合规范
    validateName(table)

    val newTableDefinition = if (tableDefinition.storage.locationUri.isDefined
      && !tableDefinition.storage.locationUri.get.isAbsolute) {
      //使表的位置合格
      val qualifiedTableLocation =
        makeQualifiedTablePath(tableDefinition.storage.locationUri.get, db)
      tableDefinition.copy(
        storage = tableDefinition.storage.copy(locationUri = Some(qualifiedTableLocation)),
        identifier = tableIdentifier)
    } else {
      tableDefinition.copy(identifier = tableIdentifier)
    }

    //会调用外部Catalog判断该数据库是否存在
    //externalCatalog.databaseExists(dbName)
    requireDbExists(db)
    //会调用外部Catalog判断该表是否存在
    //externalCatalog.tableExists(db, table)
    if (tableExists(newTableDefinition.identifier)) {
      if (!ignoreIfExists) {
        throw new TableAlreadyExistsException(db = db, table = table)
      }
    } else if (validateLocation) {
      validateTableLocation(newTableDefinition)
    }
    //还是会调用外部Catalog去创建表
    externalCatalog.createTable(newTableDefinition, ignoreIfExists)
  }

临时表

我们也以创建临时视图为例来看下内部逻辑

  //临时视图列表，从表名映射到其逻辑计划
  @GuardedBy("this")
  protected val tempViews = new mutable.HashMap[String, TemporaryViewRelation]

  def createTempView(
      name: String,
      viewDefinition: TemporaryViewRelation,
      overrideIfExists: Boolean): Unit = synchronized {
    //格式化表名
    val table = formatTableName(name)
    //判断之前是否创建过临时视图，且是否可以覆盖
    if (tempViews.contains(table) && !overrideIfExists) {
      throw new TempTableAlreadyExistsException(name)
    }
    //想map中放入该表和该表的临时视图信息（用于后续的分析计划）
    tempViews.put(table, viewDefinition)
  }

临时表就是在内存中维护了一个map来存储它，外部表和数据库都是通过ExternalCatalog来操作的，下面我们就看下ExternalCatalog

4、ExternalCatalog

系统目录（包括函数、分区、表和数据库）的接口

这仅用于非临时项，实现必须是线程安全的，因为它们可以在多个线程中访问。这是一个外部Catalog，因为它需要与外部系统交互。

它相应的也有对数据库、表、分区、函数的操作，来支撑上层的调用

它有个子类：HiveExternalCatalog（使用Hive的系统目录的持久实现）

HiveExternalCatalog里面也有对数据库、表、分区、函数的操作，并把这些操作转交给HiveClient来操作。

private[spark] class HiveExternalCatalog(conf: SparkConf, hadoopConf: Configuration)
  extends ExternalCatalog with Logging {

  org.apache.hadoop.util.VersionInfo.getVersion()

  //用于与元数据交互的Hive客户端
  lazy val client: HiveClient = {
    HiveUtils.newClientForMetadata(conf, hadoopConf)
  }

  //下面是对数据库、表、分区、函数的操作

}

四、总结：

从源码的调用，我们可以清楚的知道Catalog对临时视图、hive库表、分区、函数进行了统一管理，其中临时视图是用要给map来维护它们的关系，hive方面的实体表是委托给HiveExternalCatalog调用HiveClient来进行操作。