alter table abc add columns (stats1 map<string,string>, stats2 map<string,string>)
我用上面的问题修改了我的表。但是在检查数据之后,我得到了两个额外列的空值。我没有得到数据。截图
fumotvh31#
级联就是解决方案。查询:
ALTER TABLE dbname.table_name ADD columns (column1 string,column2 string) CASCADE;
这将更改表元数据的列,并将相同的更改级联到所有分区元数据。 RESTRICT 是默认值,将列更改仅限于表元数据。
RESTRICT
ou6hu8tu2#
正如其他人所指出的那样 CASCADE 将更改所有分区的元数据。没有 CASCADE ,如果要更改旧分区以包含新列,则需要 DROP 先把旧分区填满, INSERT OVERWRITE 没有 DROP 不会工作,因为元数据不会更新为新的默认元数据。假设你已经跑了 alter table add columns (stats1 map<string,string>, stats2 map<string,string>) 没有 CASCADE 一个意外然后你 INSERT OVERWRITE 一个没有先掉下来的旧分区。数据将存储在底层文件中,但是如果从配置单元中查询该分区的表,它将不会显示,因为元数据没有更新。无需使用以下方法重新运行insert overwrite即可修复此问题:跑 SHOW CREATE TABLE dbname.tblname 并复制添加新列之前存在的所有列定义跑 ALTER TABLE dbname.tblname REPLACE COLUMNS ({paste in col defs besides columns to add here}) CASCADE 跑 ALTER TABLE dbname.tblname ADD COLUMNS (newcol1 int COMMENT "new col") CASCADE 很高兴所有分区的元数据都已更改()作为步骤2-3的示例:
CASCADE
DROP
INSERT OVERWRITE
alter table add columns (stats1 map<string,string>, stats2 map<string,string>)
SHOW CREATE TABLE dbname.tblname
ALTER TABLE dbname.tblname REPLACE COLUMNS ({paste in col defs besides columns to add here}) CASCADE
ALTER TABLE dbname.tblname ADD COLUMNS (newcol1 int COMMENT "new col") CASCADE
DROP TABLE IF EXISTS junk.testcascade ;CREATE TABLE junk.testcascade (startcol INT)partitioned by (d int)stored as parquet;INSERT INTO TABLE junk.testcascade PARTITION(d=1)VALUES (1), (2);INSERT INTO TABLE junk.testcascade PARTITION(d=2)VALUES (1), (2);SELECT * FROM junk.testcascade ;+-----------------------+----------------+--+| testcascade.startcol | testcascade.d |+-----------------------+----------------+--+| 1 | 1 || 2 | 1 || 1 | 2 || 2 | 2 |+-----------------------+----------------+--+ --no cascade! oppsALTER TABLE junk.testcascade ADD COLUMNS( testcol1 int, testcol2 int) ;INSERT OVERWRITE TABLE junk.testcascade PARTITION(d=3)VALUES (1,1,1), (2,1,1);INSERT OVERWRITE TABLE junk.testcascade PARTITION(d=2)VALUES (1,1,1), (2,1,1);--okay! because we created this table after altering the metadataselect * FROM junk.testcascade where d=3;+-----------------------+-----------------------+-----------------------+----------------+--+| testcascade.startcol | testcascade.testcol1 | testcascade.testcol2 | testcascade.d |+-----------------------+-----------------------+-----------------------+----------------+--+| 1 | 1 | 1 | 3 || 2 | 1 | 1 | 3 |+-----------------------+-----------------------+-----------------------+----------------+--+--not okay even tho we inserted =( because the metadata isnt changedselect * FROM junk.testcascade where d=2;+-----------------------+-----------------------+-----------------------+----------------+--+| testcascade.startcol | testcascade.testcol1 | testcascade.testcol2 | testcascade.d |+-----------------------+-----------------------+-----------------------+----------------+--+| 1 | NULL | NULL | 2 || 2 | NULL | NULL | 2 |+-----------------------+-----------------------+-----------------------+----------------+--+--cut back to original columnsALTER TABLE junk.testcascade REPLACE COLUMNS( startcol int) CASCADE;--addALTER table junk.testcascade ADD COLUMNS( testcol1 int, testcol2 int) CASCADE;--it works!select * FROM junk.testcascade where d=2; +-----------------------+-----------------------+-----------------------+----------------+--+| testcascade.startcol | testcascade.testcol1 | testcascade.testcol2 | testcascade.d |+-----------------------+-----------------------+-----------------------+----------------+--+| 1 | 1 | 1 | 2 || 2 | 1 | 1 | 2 |+-----------------------+-----------------------+-----------------------+----------------+--+
DROP TABLE IF EXISTS junk.testcascade ;
CREATE TABLE junk.testcascade (
startcol INT
)
partitioned by (d int)
stored as parquet
;
INSERT INTO TABLE junk.testcascade PARTITION(d=1)
VALUES
(1),
(2)
INSERT INTO TABLE junk.testcascade PARTITION(d=2)
SELECT * FROM junk.testcascade ;
+-----------------------+----------------+--+
| testcascade.startcol | testcascade.d |
| 1 | 1 |
| 2 | 1 |
| 1 | 2 |
| 2 | 2 |
--no cascade! opps
ALTER TABLE junk.testcascade ADD COLUMNS( testcol1 int, testcol2 int) ;
INSERT OVERWRITE TABLE junk.testcascade PARTITION(d=3)
(1,1,1),
(2,1,1)
INSERT OVERWRITE TABLE junk.testcascade PARTITION(d=2)
--okay! because we created this table after altering the metadata
select * FROM junk.testcascade where d=3;
+-----------------------+-----------------------+-----------------------+----------------+--+
| testcascade.startcol | testcascade.testcol1 | testcascade.testcol2 | testcascade.d |
| 1 | 1 | 1 | 3 |
| 2 | 1 | 1 | 3 |
--not okay even tho we inserted =( because the metadata isnt changed
select * FROM junk.testcascade where d=2;
| 1 | NULL | NULL | 2 |
| 2 | NULL | NULL | 2 |
--cut back to original columns
ALTER TABLE junk.testcascade REPLACE COLUMNS( startcol int) CASCADE;
--add
ALTER table junk.testcascade ADD COLUMNS( testcol1 int, testcol2 int) CASCADE;
--it works!
| 1 | 1 | 1 | 2 |
| 2 | 1 | 1 | 2 |
h79rfbju3#
只有当您的数据被分区并且您知道最新分区的位置时,此解决方案才有效。在这种情况下,您可以执行以下操作,而不是执行代价高昂的恢复分区或修复操作:读取分区表并获取架构详细信息读取要更新的表现在找出哪些列是不同的,并为每个列做一个alter表发布scala代码以供参考:
def updateMetastoreColumns(spark: SparkSession, partitionedTablePath: String, toUpdateTableName: String): Unit = { //fetch all column names along with their corresponding datatypes from latest partition val partitionedTable = spark.read.orc(partitionedTablePath) val partitionedTableColumns = partitionedTable.columns zip partitionedTable.schema.map(_.dataType.catalogString) //fetch all column names along with their corresponding datatypes from currentTable val toUpdateTable = spark.read.table(toUpdateTableName) val toUpdateTableColumns = toUpdateTable.columns zip toUpdateTable.schema.map(_.dataType.catalogString) //check if new columns are present in newer partition val diffColumns = partitionedTableColumns.diff(toUpdateTableColumns) //update the metastore with new column info diffColumns.foreach {column: (String, String) => { spark.sql(s"ALTER TABLE ${toUpdateTableName} ADD COLUMNS (${column._1} ${column._2})") }} }
def updateMetastoreColumns(spark: SparkSession, partitionedTablePath: String, toUpdateTableName: String): Unit = {
//fetch all column names along with their corresponding datatypes from latest partition
val partitionedTable = spark.read.orc(partitionedTablePath)
val partitionedTableColumns = partitionedTable.columns zip partitionedTable.schema.map(_.dataType.catalogString)
//fetch all column names along with their corresponding datatypes from currentTable
val toUpdateTable = spark.read.table(toUpdateTableName)
val toUpdateTableColumns = toUpdateTable.columns zip toUpdateTable.schema.map(_.dataType.catalogString)
//check if new columns are present in newer partition
val diffColumns = partitionedTableColumns.diff(toUpdateTableColumns)
//update the metastore with new column info
diffColumns.foreach {column: (String, String) => {
spark.sql(s"ALTER TABLE ${toUpdateTableName} ADD COLUMNS (${column._1} ${column._2})")
}}
}
这将帮助您动态地查找添加到较新分区的最新列,并动态地将其更新到元存储中。
0x6upsns4#
要将列添加到分区表中,需要重新创建分区。假设表是外部的,并且数据文件已经包含新列,请执行以下操作:1。更改表添加列。。。2.重新创建分区。对于每个分区,先删除然后创建。新创建的分区架构将继承表架构。或者,您可以删除表,然后创建表并创建所有分区,或者简单地恢复它们 MSCK REPAIR TABLE 命令。amazon elastic mapreduce(emr)版本的hive上的等效命令是: ALTER TABLE table_name RECOVER PARTITIONS . 请参阅此处的手册:恢复分区在Hive1.1.0及更高版本中,您也可以使用 CASCADE 选择 ALTER TABLE ADD|REPLACE COLUMNS . 请参阅此处的手册:添加列这些建议适用于外部表。
MSCK REPAIR TABLE
ALTER TABLE table_name RECOVER PARTITIONS
ALTER TABLE ADD|REPLACE COLUMNS
4条答案
按热度按时间fumotvh31#
级联就是解决方案。
查询:
这将更改表元数据的列,并将相同的更改级联到所有分区元数据。
RESTRICT
是默认值,将列更改仅限于表元数据。ou6hu8tu2#
正如其他人所指出的那样
CASCADE
将更改所有分区的元数据。没有CASCADE
,如果要更改旧分区以包含新列,则需要DROP
先把旧分区填满,INSERT OVERWRITE
没有DROP
不会工作,因为元数据不会更新为新的默认元数据。假设你已经跑了
alter table add columns (stats1 map<string,string>, stats2 map<string,string>)
没有CASCADE
一个意外然后你INSERT OVERWRITE
一个没有先掉下来的旧分区。数据将存储在底层文件中,但是如果从配置单元中查询该分区的表,它将不会显示,因为元数据没有更新。无需使用以下方法重新运行insert overwrite即可修复此问题:跑
SHOW CREATE TABLE dbname.tblname
并复制添加新列之前存在的所有列定义跑
ALTER TABLE dbname.tblname REPLACE COLUMNS ({paste in col defs besides columns to add here}) CASCADE
跑ALTER TABLE dbname.tblname ADD COLUMNS (newcol1 int COMMENT "new col") CASCADE
很高兴所有分区的元数据都已更改()作为步骤2-3的示例:
h79rfbju3#
只有当您的数据被分区并且您知道最新分区的位置时,此解决方案才有效。在这种情况下,您可以执行以下操作,而不是执行代价高昂的恢复分区或修复操作:
读取分区表并获取架构详细信息
读取要更新的表
现在找出哪些列是不同的,并为每个列做一个alter表
发布scala代码以供参考:
这将帮助您动态地查找添加到较新分区的最新列,并动态地将其更新到元存储中。
0x6upsns4#
要将列添加到分区表中,需要重新创建分区。假设表是外部的,并且数据文件已经包含新列,请执行以下操作:1。更改表添加列。。。2.重新创建分区。对于每个分区,先删除然后创建。新创建的分区架构将继承表架构。
或者,您可以删除表,然后创建表并创建所有分区,或者简单地恢复它们
MSCK REPAIR TABLE
命令。amazon elastic mapreduce(emr)版本的hive上的等效命令是:ALTER TABLE table_name RECOVER PARTITIONS
. 请参阅此处的手册:恢复分区在Hive1.1.0及更高版本中,您也可以使用
CASCADE
选择ALTER TABLE ADD|REPLACE COLUMNS
. 请参阅此处的手册:添加列这些建议适用于外部表。