title: How to Alter Table from SparkSQL based on the DataFrame without dropping the hive table, like drop/add columns?
body: <p>I am Trying to Alter a Hive Table using Spark, like adding columns or dropping columns from the Hive table based on the Spark DataFrame output. Below is what I tried, kind of a huge code,</p>
  
  <pre><code>    def main(args: Array[String]): Unit = {
      implicit val spark = SparkSession.builder
        .appName("SchemaHandle")
        .enableHiveSupport
        .getOrCreate
  
      //Assume below is my generated DataFrame
      import spark.implicits._
      val dfSample = Seq(
        (12, "Dallas", "Texas", 55, "BOOK S","hello","Hellotwo"),
        (12, "SF", "CA", 25, "RULER","hello","Hellotwo"),
        (13, "NYC", "NY", 53, "PENCIL S","hello","Hellotwo"),
        (14, "Miami", "Fl", 45, "RULER","hello","Hellotwo"),
        (12, "Houston", "Texas", 75, "MARKER","hello","Hellotwo"),
        (11, "jersey", "NJ", 53, "WHITE NE R","hello","Hellotwo"),
        (19, "new orleans", "LO", 45, "HIGHLIGHTNER","hello","Hellotwo")
      ).toDF("id", "city", "state", "qty", "item","columnone","columntwo")
  
      try {
        spark.sql("truncate table database.schematest")
        println("Successfully truncated database.schematest")
      } catch {
        case _: Throwable =&gt; println("This Job is running for the very first time, so no table to truncate - We'll create the table below")
          dfSample.write.format("parquet").mode(SaveMode.Overwrite).saveAsTable(s"database.schematest")
          println("Output Table Saved to database.schematest")
      }
  
      //Assume this is Spark DF Schema.
      val seqone: Seq[StructField] = dfSample.schema
      //Assume this is Existing Table Schema.
      val seqtwo: Seq[StructField] = spark.table("database.schematest").schema
  
      //Get Cols- with Schema to be Added
      val diffedSeq = seqone diff seqtwo
      //Get Cols- with Schema to be Dropped
      val diffedSeqTwo = seqtwo diff seqone
  
      //Get Cols- names to just make the diff
      val seqonecolumns = dfSample.columns
  
      //Get Cols- names to just make the diff
      val seqtwocolumns = spark.table("dscoewrk_ing_qa.schematest").columns
  
      val diffedSeqArrayOne = seqonecolumns diff seqtwocolumns
  
      val diffedSeqArrayTwo = seqtwocolumns diff seqonecolumns
  
      var fixedAlterColumns: String = ""
      for (i &lt;- diffedSeqArrayOne) {
        for (j &lt;- diffedSeq) {
          if (i.equals(j.name)) {
            fixedAlterColumns +=""+j.name +" "+ datatypeCheckFunction(j.dataType.toString)+","
          }
        }
      }
      if(fixedAlterColumns.length&gt;0) {
        println(s"Result---&gt; ${fixedAlterColumns.substring(0, fixedAlterColumns.length - 1)}")
        //Lets add new columns to table database.schematest.
        spark.sql(s"ALTER TABLE database.schematest ADD COLUMNS (${fixedAlterColumns.substring(0, fixedAlterColumns.length - 1)})")
        println("Alter Table Success")
      }else{
        println("No Columns to Add")
      }
  
      println("------------------------------BREAK---------------------------")
  
      //Now lets think about dropping the columns
      val dfSampleCurrentTable:Seq[StructField] = spark.table("dscoewrk_ing_qa.schematest").schema
      //Since we cannot drop columns from Hive Table, lets do REPLACE COLUMNS.
      val dfSampleFinalDiff = dfSampleCurrentTable diff diffedSeqTwo
      dfSampleFinalDiff.foreach(println)
      val dfSampleFinalDiffColArray = (spark.table("database.schematest").columns) diff diffedSeqArrayTwo
      dfSampleFinalDiffColArray.foreach(println)
      var fixedDropColumns:String = ""
  
      for(i &lt;- dfSampleFinalDiffColArray){
        println("The i is"+i)
        for(j &lt;-dfSampleCurrentTable){
          println("This is j"+j)
          if(i.equals(j.name)){
            fixedDropColumns+=""+j.name +" "+ datatypeCheckFunction(j.dataType.toString)+","
          }
        }
      }
      //Let's drop the columns that aren't required.
  
      if(fixedDropColumns.length&gt;0) {
        println(s"Result---&gt; ${fixedDropColumns.substring(0, fixedDropColumns.length - 1)}")
        spark.sql(s"ALTER TABLE database.schematest REPLACE COLUMNS(${fixedDropColumns.substring(0,fixedDropColumns.length-1)})")
        println("Alter Drop Table Success")
      }else{
        println("No Columns to Drop")
      }
  
      //Now let's save the DF to Output in the Table. By using Append as below.
      dfSample.withColumn("mybool",functions.lit(null)).coalesce(50).write.format("parquet").mode(SaveMode.Append).insertInto("database.schematest")
  
      println("Saving output Table Successful.")
  
    }
  
     def datatypeCheckFunction(datatypePassed: String): String = {
      datatypePassed match {
        case "BinaryType" | "ByteType" | "DateType" | "NullType" | "StringType" | "TimestampType" =&gt; "String"
        case "BooleanType" =&gt; "boolean"
        case "DoubleType" | "FloatType" =&gt; "Double"
        case "IntegerType" | "ShortType" =&gt; "Int"
        case "LongType" =&gt; "BigInt"
        case _ =&gt; "String"
      }
    }
  }
  </code></pre>
  
  <p>I can understand that there is room for optimization but atleast for this code, I see two issues:
   1. When I run the above Spark job, Add columns is successfull but the REPLACE columns is failing with below:</p>
  
  <pre><code>Exception in thread "main" org.apache.spark.sql.catalyst.parser.ParseException:
  Operation not allowed: ALTER TABLE REPLACE COLUMNS(line 1, pos 0)
  
  == SQL ==
  ALTER TABLE database.schematest REPLACE COLUMNS(id Int,city String,state String,qty Int,item String,columnone String,columntwo String)
  
  </code></pre>
  
  <ol start="2">
  <li>Assume REPLACE Columns has worked, will it also delete the data for that dropping column?</li>
  </ol>
  
  <p>Here is my Hive Create Table Statement used:</p>
  
  <pre><code>create table schematest(`id` int, `city` string, `state` string, `qty` int, `mybool` boolean) stored as parquet
  </code></pre>
  
  <p>Any help is appreciated, Thank you All in Advance.</p>