Spark Learning: implement compact table command

requirement : Add add compact table command , Used to merge small files

One 、 modify SqlBase.g4

#statement Add below 
| COMPACT TABLE target=tableIdentifier partitionSpec?
(INTO fileNum=INTEGER_VALUE FILES)?                        #compactTable
#ansiNonReserved Add below 
| FILES
#nonReserved Add below 
| FILES
--keywords list Add below 
FILES:'FILES';

Two 、maven compile

3、 ... and 、 modify SparkSqlParser.scala

 override def visitCompactTable(
                                  ctx: CompactTableContext): LogicalPlan = withOrigin(ctx) {
    
    val table = visitTableIdentifier(ctx.tableIdentifier())
    val filesNum = if (ctx.INTEGER_VALUE() != null) {
    
      Some(ctx.INTEGER_VALUE().getText)
    } else {
    
      None
    }
    val partition = if (ctx.partitionSpec() != null) {
    
      Some(ctx.partitionSpec().getText)
    } else {
    
      None
    }
    CompactTableCommand(table,filesNum,partition);
  }

Four 、 increase compactTableCommand class

package org.apache.spark.sql.execution.command

import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.expressions.{
    Attribute, AttributeReference}
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.{
    Row, SaveMode, SparkSession}

case class CompactTableCommand(table: TableIdentifier,
                               filesNum: Option[String],
                               partitionSpec: Option[String]) extends LeafRunnableCommand {
    

  private val defaultSize = 128 * 1024 * 1024

  override def output: Seq[Attribute] = Seq(
    AttributeReference("COMPACT_TABLE", StringType, nullable = false)()
  )

  override def run(sparkSession: SparkSession): Seq[Row] = {
    
    sparkSession.catalog.setCurrentDatabase(table.database.getOrElse("default"))


    val tempTableName = "`" + table.identifier + "_" + System.currentTimeMillis() + "`"

    val originDataFrame = sparkSession.table(table.identifier)
    val partitions = filesNum match {
    
      case Some(files) => files.toInt
      case None => (sparkSession.sessionState
        .executePlan(originDataFrame.queryExecution.logical)
        .optimizedPlan.stats.sizeInBytes / defaultSize).toInt + 1
    }
    // scalastyle:off println
    println(partitions, tempTableName)

    if (partitionSpec.nonEmpty) {
    
      // https://stackoverflow.com/questions/38487667/
      // overwrite-specific-partitions-in-spark-dataframe-write-method
      sparkSession.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

      val conditionExpr = partitionSpec.get.trim.stripPrefix("partition(").dropRight(1)
        .replace(",", "AND")
      // scalastyle:off println
      println(conditionExpr)

      originDataFrame.where(conditionExpr).repartition(partitions)
        .write
        .mode(SaveMode.Overwrite)
        .saveAsTable(tempTableName)

      sparkSession.table(tempTableName).write
        .mode(SaveMode.Overwrite)
        .insertInto(table.identifier)
    } else {
    
      //
      originDataFrame.repartition(partitions)
        .write
        .mode(SaveMode.Overwrite)
        .saveAsTable(tempTableName)

      sparkSession.table(tempTableName)
        .write
        .mode(SaveMode.Overwrite)
        .saveAsTable(table.identifier)
    }

    // sparkSession.sql(s"DROP TABLE ${tempTableName}")

    Seq(Row(s"compact table ${
      table.identifier} finished."))
  }
}

5、 ... and 、 Report errors

1、 Recompile to make it effective

mvn package -DskipTests -Phive -Phive-thriftserver

2、 Create test data

#spark-sql Create a table 
create table if not exists orders
(
    id int COMMENT 'order id'
    ,name string COMMENT 'user name'
);
# insert data 
INSERT INTO orders
VALUES 
  (1, 'part-1'), 
  (2, 'part-2'),
  (3, 'part-3'), 
  (4, 'part-4'),
  (5, 'part-5'), 
  (6, 'part-6');

3、 Report errors

This error report took a day or two , It's useless to refer to online solutions , Finally, we reconfigured a set hadoop Environmental Science
1) download hadoop, I downloaded it 3.3.2
hadoop Mirror address
2) To configure HADOOP_HOME To the directory just extracted
3) Download the corresponding version of winutils.exe and hadoop.dll Put it in the decompressed bin Catalog , And put hadoop.dll Put it in System32 Under the table of contents
hadoop.dll、winutils.exe Download address
4) Reinsert data

6、 ... and 、 verification

newly build orders surface , Insert 6 Data , You can see that there is 6 File

Use compact table command command

compact table orders into 3 Files;

As shown in the figure below , You can see that it becomes three files , And a backup table is added