模式演进

-- add columns
ALTER TABLE tableName ADD COLUMNS(col_spec[, col_spec ...])

ALTER TABLE h0 ADD COLUMNS(ext0 string);
ALTER TABLE h0 ADD COLUMNS(new_col int not null comment 'add new column' AFTER col1);
ALTER TABLE complex_table ADD COLUMNS(col_struct.col_name string comment 'add new column to a struct col' AFTER col_from_col_struct);

-- alter table ... alter column
ALTER TABLE tableName ALTER [COLUMN] col_old_name TYPE column_type [COMMENT] col_comment[FIRST|AFTER] column_name

--- Changing the column type
ALTER TABLE table1 ALTER COLUMN a.b.c TYPE bigint

--- Altering other attributes
ALTER TABLE table1 ALTER COLUMN a.b.c COMMENT 'new comment'
ALTER TABLE table1 ALTER COLUMN a.b.c FIRST
ALTER TABLE table1 ALTER COLUMN a.b.c AFTER x
ALTER TABLE table1 ALTER COLUMN a.b.c DROP NOT NULL

-- alter table ... drop columns
ALTER TABLE tableName DROP COLUMN|COLUMNS cols

ALTER TABLE table1 DROP COLUMN a.b.c
ALTER TABLE table1 DROP COLUMNS a.b.c, x, y

-- alter table ... rename column
ALTER TABLE tableName RENAME COLUMN old_columnName TO new_columnName

ALTER TABLE table1 RENAME COLUMN a.b.c TO x

--conf 'spark.hadoop.hive.metastore.disallow.incompatible.col.type.changes=false'

<property>
  <name>hive.metastore.disallow.incompatible.col.type.changes</name>
  <value>false</value>
</property>

set metaconf:hive.metastore.disallow.incompatible.col.type.changes=false;

scala> :paste
import org.apache.hudi.QuickstartUtils._
import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row

val tableName = "hudi_trips_cow"
val basePath = "file:///tmp/hudi_trips_cow"
val schema = StructType( Array(
     StructField("rowId", StringType,true),
     StructField("partitionId", StringType,true),
     StructField("orderingField", LongType,true),
     StructField("name", StringType,true),
     StructField("versionId", StringType,true),
     StructField("intToLong", IntegerType,true)
))


val data1 = Seq(Row("row_1", "part_0", 0L, "bob", "v_0", 0),
                Row("row_2", "part_0", 0L, "john", "v_0", 0),
                Row("row_3", "part_0", 0L, "tom", "v_0", 0))

var dfFromData1 = spark.createDataFrame(data1, schema)
dfFromData1.write.format("hudi").
   options(getQuickstartWriteConfigs).
   option("hoodie.table.ordering.fields", "orderingField").
   option("hoodie.datasource.write.recordkey.field", "rowId").
   option("hoodie.datasource.write.partitionpath.field", "partitionId").
   option("hoodie.index.type","SIMPLE").
   option("hoodie.table.name", tableName).
   mode(Overwrite).
   save(basePath)

var tripsSnapshotDF1 = spark.read.format("hudi").load(basePath + "/*/*")
tripsSnapshotDF1.createOrReplaceTempView("hudi_trips_snapshot")

ctrl+D

scala> spark.sql("desc hudi_trips_snapshot").show()
    +--------------------+---------+-------+
    |            col_name|data_type|comment|
    +--------------------+---------+-------+
    | _hoodie_commit_time|   string|   null|
    |_hoodie_commit_seqno|   string|   null|
    |  _hoodie_record_key|   string|   null|
    |_hoodie_partition...|   string|   null|
    |   _hoodie_file_name|   string|   null|
    |               rowId|   string|   null|
    |         partitionId|   string|   null|
    |       orderingField|   bigint|   null|
    |                name|   string|   null|
    |           versionId|   string|   null|
    |           intToLong|      int|   null|
    +--------------------+---------+-------+

scala> spark.sql("select rowId, partitionId, orderingField, name, versionId, intToLong from hudi_trips_snapshot").show()
    +-----+-----------+-------------+----+---------+---------+
    |rowId|partitionId|orderingField|name|versionId|intToLong|
    +-----+-----------+-------------+----+---------+---------+
    |row_3|     part_0|            0| tom|      v_0|        0|
    |row_2|     part_0|            0|john|      v_0|        0|
    |row_1|     part_0|            0| bob|      v_0|        0|
    +-----+-----------+-------------+----+---------+---------+

// In the new schema, we are going to add a String field and
// change the datatype `intToLong` field from  int to long.
scala> :paste
val newSchema = StructType( Array(
    StructField("rowId", StringType,true),
    StructField("partitionId", StringType,true),
    StructField("orderingField", LongType,true),
    StructField("name", StringType,true),
    StructField("versionId", StringType,true),
    StructField("intToLong", LongType,true),
    StructField("newField", StringType,true)
))

val data2 = Seq(Row("row_2", "part_0", 5L, "john", "v_3", 3L, "newField_1"),
               Row("row_5", "part_0", 5L, "maroon", "v_2", 2L, "newField_1"),
               Row("row_9", "part_0", 5L, "michael", "v_2", 2L, "newField_1"))

var dfFromData2 = spark.createDataFrame(data2, newSchema)
dfFromData2.write.format("hudi").
    options(getQuickstartWriteConfigs).
    option("hoodie.table.ordering.fields", "orderingField").
    option("hoodie.datasource.write.recordkey.field", "rowId").
    option("hoodie.datasource.write.partitionpath.field", "partitionId").
    option("hoodie.index.type","SIMPLE").
    option("hoodie.table.name", tableName).
    mode(Append).
    save(basePath)

var tripsSnapshotDF2 = spark.read.format("hudi").load(basePath + "/*/*")
tripsSnapshotDF2.createOrReplaceTempView("hudi_trips_snapshot")

Ctrl + D

scala> spark.sql("desc hudi_trips_snapshot").show()
    +--------------------+---------+-------+
    |            col_name|data_type|comment|
    +--------------------+---------+-------+
    | _hoodie_commit_time|   string|   null|
    |_hoodie_commit_seqno|   string|   null|
    |  _hoodie_record_key|   string|   null|
    |_hoodie_partition...|   string|   null|
    |   _hoodie_file_name|   string|   null|
    |               rowId|   string|   null|
    |         partitionId|   string|   null|
    |       orderingField|   bigint|   null|
    |                name|   string|   null|
    |           versionId|   string|   null|
    |           intToLong|   bigint|   null|
    |            newField|   string|   null|
    +--------------------+---------+-------+


scala> spark.sql("select rowId, partitionId, orderingField, name, versionId, intToLong, newField from hudi_trips_snapshot").show()
    +-----+-----------+-------------+-------+---------+---------+----------+
    |rowId|partitionId|orderingField|   name|versionId|intToLong|  newField|
    +-----+-----------+-------------+-------+---------+---------+----------+
    |row_3|     part_0|            0|    tom|      v_0|        0|      null|
    |row_2|     part_0|            5|   john|      v_3|        3|newField_1|
    |row_1|     part_0|            0|    bob|      v_0|        0|      null|
    |row_5|     part_0|            5| maroon|      v_2|        2|newField_1|
    |row_9|     part_0|            5|michael|      v_2|        2|newField_1|
    +-----+-----------+-------------+-------+---------+---------+----------+