前言

Spark SQL 在刪除外部表時，本不能刪除外部表的資料的。本篇文章主要介紹如何修改Spark SQL 原始碼實現在刪除外部表的時候，可以帶額外選項來刪除外部表的資料。

本文的環境是我一直使用的 spark 2.4.3 版本。

1. 修改ANTLR4 語法檔案

修改 SqlBase.g4檔案中drop Table 相關語句，新增(WITH DATA)?， 修改完之後如下：

DROP TABLE (IF EXISTS)? tableIdentifier (WITH DATA)? PURGE?                   #dropTable

因為，刪除external表也不是必須的，所以新增WITH DATA 為可選項，跟 IF EXISTS類似。

2. 修改相關方法

2.1 修改SparkSqlParser.scala檔案

  /**
   * Create a [[DropTableCommand]] command.
   */
  override def visitDropTable(ctx: DropTableContext): LogicalPlan = withOrigin(ctx) {
    DropTableCommand(
      visitTableIdentifier(ctx.tableIdentifier),
      ctx.EXISTS != null,
      ctx.VIEW != null,
      ctx.PURGE != null,
      ctx.WITH() != null && ctx.DATA() != null)
  }

2.2 修改DropTableCommand.scala等相關檔案

首先修改建構函式，在最後一個引數後面新增withData方法，預設為false：

case class DropTableCommand(
    tableName: TableIdentifier,
    ifExists: Boolean,
    isView: Boolean,
    purge: Boolean,
    withData:Boolean = false // TODO 外部表是否需要刪除表資料
    ) extends RunnableCommand

DropTableCommand本質上其實是用了command設計模式，實際在執行時，會呼叫其run方法，修改 run 方法，如下：

 1 override def run(sparkSession: SparkSession): Seq[Row] = {
 2     val catalog = sparkSession.sessionState.catalog
 3     val isTempView = catalog.isTemporaryTable(tableName)
 4 
 5     if (!isTempView && catalog.tableExists(tableName)) {
 6       // If the command DROP VIEW is to drop a table or DROP TABLE is to drop a view
 7       // issue an exception.
 8       catalog.getTableMetadata(tableName).tableType match {
 9         case CatalogTableType.VIEW if !isView =>
10           throw new AnalysisException(
11             "Cannot drop a view with DROP TABLE. Please use DROP VIEW instead")
12         case o if o != CatalogTableType.VIEW && isView =>
13           throw new AnalysisException(
14             s"Cannot drop a table with DROP VIEW. Please use DROP TABLE instead")
15         case _ =>
16       }
17     }
18 
19     if (isTempView || catalog.tableExists(tableName)) {
20       try {
21         sparkSession.sharedState.cacheManager.uncacheQuery(
22           sparkSession.table(tableName), cascade = !isTempView)
23       } catch {
24         case NonFatal(e) => log.warn(e.toString, e)
25       }
26       catalog.refreshTable(tableName)
27       log.warn(s"withData:${withData}")
28       catalog.dropTable(tableName, ifExists, purge, withData)
29     } else if (ifExists) {
30       // no-op
31     } else {
32       throw new AnalysisException(s"Table or view not found: ${tableName.identifier}")
33     }
34     Seq.empty[Row]
35   }

在第 28 行，為 catalog物件的dropTable 新增 withData 引數。其中catalog是 org.apache.spark.sql.catalyst.catalog.SessionCatalog 的例項。其子類並沒有重寫其 dropTable 方法，故只需要修改其dropTable 方法即可。具體修改程式碼如下：

 1 /**
 2    * Drop a table.
 3    *
 4    * If a database is specified in `name`, this will drop the table from that database.
 5    * If no database is specified, this will first attempt to drop a temporary view with
 6    * the same name, then, if that does not exist, drop the table from the current database.
 7    */
 8   def dropTable(
 9       name: TableIdentifier,
10       ignoreIfNotExists: Boolean,
11       purge: Boolean,
12       withData:Boolean = false // 外部表是否需要在hdfs上刪除其對應的資料
13                ): Unit = synchronized {
14     val db = formatDatabaseName(name.database.getOrElse(currentDb))
15     val table = formatTableName(name.table)
16     if (db == globalTempViewManager.database) {
17       val viewExists = globalTempViewManager.remove(table)
18       if (!viewExists && !ignoreIfNotExists) {
19         throw new NoSuchTableException(globalTempViewManager.database, table)
20       }
21     } else {
22       if (name.database.isDefined || !tempViews.contains(table)) {
23         requireDbExists(db)
24         // When ignoreIfNotExists is false, no exception is issued when the table does not exist.
25         // Instead, log it as an error message.
26         if (tableExists(TableIdentifier(table, Option(db)))) {
27           logError(s"withData :${withData}")
28           externalCatalog.dropTable(db, table, ignoreIfNotExists = true,purge = purge, withData)
29         } else if (!ignoreIfNotExists) {
30           throw new NoSuchTableException(db = db, table = table)
31         }
32       } else {
33         tempViews.remove(table)
34       }
35     }
36   }

為防止在test中有很多的測試類在呼叫該方法，在編譯時報錯，新新增的withData給預設值，為false，保證該方法預設行為跟之前未修改前一致。

withData 引數繼續傳遞給 externalCatalog.dropTable 方法，其中，externalCatalog 是 org.apache.spark.sql.catalyst.catalog.ExternalCatalog 型別變數，ExternalCatalog 是一個trait，ExternalCatalog 實現類關係如下：

 首先修改ExternalCatalog 的dropTable 方法，如下：

def dropTable(
      db: String,
      table: String,
      ignoreIfNotExists: Boolean,
      purge: Boolean,
      withData:Boolean=false): Unit

引數載入最後，給預設值false。

org.apache.spark.sql.catalyst.catalog.ExternalCatalogWithListener 是一個包裝類，其內部在原來ExternalCatalog 的行為之外添加了監聽的行為。先修改這個包裝類的dropTable，如下：

override def dropTable(
      db: String,
      table: String,
      ignoreIfNotExists: Boolean,
      purge: Boolean,
      withData:Boolean): Unit = {
    postToAll(DropTablePreEvent(db, table))
    delegate.dropTable(db, table, ignoreIfNotExists, purge, withData)
    postToAll(DropTableEvent(db, table))
  }

其中，delegate 就是真正執行 dropTable操作的ExternalCatalog物件。

catlog有兩個來源，分別是 in-memory和 hive， in-memory的實現類是org.apache.spark.sql.catalyst.catalog.InMemoryCatalog，只需要新增 方法引數列表即可，在方法內部不需要做任何操作。

hive的實現類是 org.apache.spark.sql.hive.HiveExternalCatalog， 其dropTable 方法如下:

override def dropTable(
      db: String,
      table: String,
      ignoreIfNotExists: Boolean,
      purge: Boolean,
      withData:Boolean): Unit = withClient {
    requireDbExists(db)
    val tableLocation: URI = client.getTable(db,table).location
    client.dropTable(db, table, ignoreIfNotExists, purge)
    val path: Path = new Path(tableLocation)
    val fileSystem: FileSystem = FileSystem.get(hadoopConf)
    val fileExists: Boolean = fileSystem.exists(path)
    logWarning(s"withData:${withData}, ${path} exists : ${fileExists}")
    if (withData && fileExists) {
      fileSystem.delete(path, true)
    }
  }

3. 打包編譯

在生產環境編譯，編譯命令如下：

./dev/make-distribution.sh --name 2.6.0-cdh5.14.0 --tgz --mvn /opt/soft/apache-maven-3.6.1/bin/mvn  -Pyarn -Phadoop-2.6 -Phive -Phive-thriftserver -Dhadoop.version=2.6.0-cdh5.14.0 -X

注：由於編譯的是 cdh版本，一些jar包不在中央倉庫，在pom.xml檔案中，新增 cloudera maven 源：

<repository>
   <id>cloudera</id>
   <url>https://repository.cloudera.com/artifactory/cloudera-repos</url>
</repository>

為了加快 maven編譯的速度， 在 make-distribution.sh 檔案中，修改了編譯的並行度，在171行，把1C改為4C，具體修改如下：

BUILD_COMMAND=("$MVN" -T 4C clean package -DskipTests $@)

執行編譯結束之後，在專案的根目錄下，會有 spark-2.4.3-bin-2.6.0-cdh5.14.0.tgz 這個壓縮包，這就是binary 檔案，可以解壓到指定目錄進行相應配置了。

4. 配置spark

把原來叢集中spark 的配置以及相關jar包拷貝到新的spark相應目錄。

5. 測試

5.1 建立外部表

spark sql

spark-sql> use test;

spark-sql> create external table ext1 location '/user/hive/warehouse/test.db/ext1' as select * from person;

spark-sql> select * from ext1;

1 2 3
2 zhangsan 4
3 lisi 5
4 wangwu 6
5 rose 7
6 nose 8
7 info 9
8 test 10

檢視 hdfs 上對應目錄是否有資料

[root@xxx ~]# hdfs dfs -ls -R /user/hive/warehouse/test.db/ext1
-rwxr-xr-x 3 root supergroup 76 2020-02-27 15:58 /user/hive/warehouse/test.db/ext1/part-00000-aae237ac-4a0b-425c-a0f1-5d54d1e88957-c000

5.2 刪除表

spark-sql> drop table if exists ext1 with data;

5.3 驗證表元資料已刪除成功

spark-sql> show tables;
test    person    false

沒有ext表，說明已刪除成功。

5.4 驗證hdfs上資料已刪除成功

[root@xxx ~]# hdfs dfs -ls -R /user/hive/warehouse/test.db/ext1
ls: `/user/hive/warehouse/test.db/ext1': No such file or directory

該目錄已不存在，說明hdfs上資料已刪除成功。

總結

本文具體介紹瞭如何修改spark sql 的原始碼，在刪除external表時可選擇地刪除hdfs上的底層數