scala - 如何向现有 Iceberg 表添加分区

Question

如何为现有的未分区的 Iceberg 表添加分区？表格已加载数据。

表已创建:

import org.apache.iceberg.hive.HiveCatalog
import org.apache.iceberg.catalog._
import org.apache.iceberg.spark.SparkSchemaUtil
import org.apache.iceberg.PartitionSpec
import org.apache.spark.sql.SaveMode._

val df1 = spark
  .range(1000)
  .toDF
  .withColumn("level",lit("something"))

val catalog = new HiveCatalog(spark.sessionState.newHadoopConf())

val icebergSchema = SparkSchemaUtil.convert(df1.schema)

val icebergTableName = TableIdentifier.of("default", "icebergTab")

val icebergTable = catalog
  .createTable(icebergTableName, icebergSchema, PartitionSpec.unpartitioned)

有什么建议吗？

Answer 1

目前，添加分区的方法是手动更新分区规范。

val table = catalog.loadTable(tableName)
val ops = table.asInstanceOf[BaseTable].operations
val spec = PartitionSpec.builderFor(table.schema).identity("level").build

val base = ops.current
val newMeta = base.updatePartitionSpec(spec)
ops.commit(base, newMeta)

有一个拉取请求来添加一个操作来进行更改，比如 addField("level")，但这还没有完全完成。我认为它会出现在 0.11.0 版本中。

请记住:

• 更改分区规范后，现有数据文件在分区字段的元数据表中将具有空值。这并不意味着如果数据是用新规范写入的，这些值会为空，只是元数据没有现有数据文件的值。
• 动态分区替换在新规范中会有不同的行为，因为分区的粒度不同。如果没有规范，INSERT OVERWRITE 将替换整个表。使用规范，只有具有新行的分区将被替换。为避免这种情况，我们建议在 Spark 中使用 DataFrameWriterV2 接口(interface)，您可以在其中更明确地说明哪些数据值被覆盖。