Create table schema from dataset schema

CreateTableStatement.scala

  import java.lang.reflect.Method
  import java.net.URI

  import org.apache.spark.sql.SparkSession
  import org.apache.spark.sql.catalyst.TableIdentifier
  import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
  import org.apache.spark.sql.execution.command.ShowCreateTableCommand
  import org.apache.spark.sql.types.StructType

  def showCreateTableCommand(
      spark: SparkSession,
      schema: StructType,
      tableName: String,
      path: String,
      inputFormat: Option[String],
      partitionColumn: Seq[String],
      properties: Map[String, String] = Map.empty[String, String]): String = {

    val identifier: TableIdentifier = spark.sessionState.sqlParser.parseTableIdentifier(tableName)
    val url: Some[URI] = Some(new URI(path))
    val outputFormat: Some[String] =
      Some("org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat")
    val hiveSerDe: Some[String] =
      Some("org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe")
    val compressed = false
    val storageFormat =
      CatalogStorageFormat(url, inputFormat, outputFormat, hiveSerDe, compressed, properties)
    val catalogTable = new CatalogTable(identifier, CatalogTableType.EXTERNAL, storageFormat,
      schema, partitionColumnNames = partitionColumn)

    val table = ShowCreateTableCommand(identifier)
    val field: Method =
      table.getClass.getDeclaredMethod("showCreateHiveTable", catalogTable.getClass)
    field.setAccessible(true)
    field.invoke(table, catalogTable).asInstanceOf[String]
  }

  val tcstmt = showCreateTableCommand(spark, df.schema, "<table name>",
    "<location of data for external table>",
    Some("org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat"), Seq("<partition column names>"))
  println(tcstmt)