数据湖（十四）：Spark与Iceberg整合查询操作

val spark: SparkSession = SparkSession.builder().master("local").appName("test")
  //指定hadoop catalog，catalog名称为hadoop_prod
  .config("spark.sql.catalog.hadoop_prod", "org.apache.iceberg.spark.SparkCatalog")
  .config("spark.sql.catalog.hadoop_prod.type", "hadoop")
  .config("spark.sql.catalog.hadoop_prod.warehouse", "hdfs://mycluster/sparkoperateiceberg")
  .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
  .getOrCreate()

//1.创建Iceberg表，并插入数据
spark.sql(
  """
    |create table hadoop_prod.mydb.mytest (id int,name string,age int) using iceberg
  """.stripMargin)
spark.sql(
  """
    |insert into hadoop_prod.mydb.mytest values (1,"zs",18),(2,"ls",19),(3,"ww",20)
  """.stripMargin)
//1.SQL 方式读取Iceberg中的数据
spark.sql("select * from hadoop_prod.mydb.mytest").show()

/**
  * 2.使用Spark查询Iceberg中的表除了使用sql 方式之外，还可以使用DataFrame方式,建议使用SQL方式
  */
//第一种方式使用DataFrame方式查询Iceberg表数据
val frame1: DataFrame = spark.table("hadoop_prod.mydb.mytest")
frame1.show()

//第二种方式使用DataFrame加载 Iceberg表数据
val frame2: DataFrame = spark.read.format("iceberg").load("hdfs://mycluster/sparkoperateiceberg/mydb/mytest")
frame2.show()

//向表 hadoop_prod.mydb.mytest 中再次插入以下数据
spark.sql(
  """
    |insert into hadoop_prod.mydb.mytest values (4,"ml",18),(5,"tq",19),(6,"gb",20)
  """.stripMargin)

//3.查看Iceberg表快照信息
spark.sql(
  """
    |select * from hadoop_prod.mydb.mytest.snapshots
  """.stripMargin).show(false)

//4.查询表历史,实际上就是表快照的部分内容
spark.sql(
  """
    |select * from hadoop_prod.mydb.mytest.history
  """.stripMargin).show(false)

//5.查看表对应的data files
spark.sql(
  """
    |select * from hadoop_prod.mydb.mytest.files
  """.stripMargin).show(false)

//6.查看表对应的 Manifests
spark.sql(
  """
    |select * from hadoop_prod.mydb.mytest.manifests
  """.stripMargin).show(false)

//7.查询指定快照数据,快照ID可以通过读取json元数据文件获取
spark.read
  .option("snapshot-id",3368002881426159310L)
  .format("iceberg")
  .load("hdfs://mycluster/sparkoperateiceberg/mydb/mytest")
  .show()

//SQL 方式指定查询快照ID 数据
spark.sql(
  """
    |call hadoop_prod.system.set_current_snapshot('mydb.mytest',3368002881426159310)
  """.stripMargin)
spark.sql(
  """
    |select * from hadoop_prod.mydb.mytest
  """.stripMargin).show()

//8.根据时间戳查询数据,时间戳指定成毫秒，iceberg会根据元数据找出timestamp-ms <= as-of-timestamp 对应的 snapshot-id ，把数据查询出来
spark.read.option("as-of-timestamp","1640066148000")
  .format("iceberg")
  .load("hdfs://mycluster/sparkoperateiceberg/mydb/mytest")
  .show()

//省略重新创建表mytest，两次插入数据
//SQL 方式查询指定 时间戳 快照数据
spark.sql(
  """
    |CALL hadoop_prod.system.rollback_to_timestamp('mydb.mytest', TIMESTAMP '2021-12-23 16:56:40.000')
  """.stripMargin)
spark.sql(
  """
    |select * from hadoop_prod.mydb.mytest
  """.stripMargin).show()

//9.回滚到某个快照，rollbackTo(snapshot-id)，指定的是固定的某个快照ID，回滚之后，会生成新的Snapshot-id, 重新查询生效。
val conf = new Configuration()
val catalog = new HadoopCatalog(conf,"hdfs://mycluster/sparkoperateiceberg")
catalog.setConf(conf)
val table: Table = catalog.loadTable(TableIdentifier.of("mydb","mytest"))
table.manageSnapshots().rollbackTo(3368002881426159310L).commit()

//查询表 hadoop_prod.mydb.mytest 数据,已经是历史数据
spark.sql(
  """
    |select * from hadoop_prod.mydb.mytest
  """.stripMargin).show(100)

//省略重新创建表mytest，两次插入数据
//SQL方式回滚快照ID，操作如下：
spark.sql(
  """
    |Call hadoop_prod.system.rollback_to_snapshot("mydb.mytest",5440886662709904549)
  """.stripMargin)

//查询表 hadoop_prod.mydb.mytest 数据,已经是历史数据
spark.sql(
  """
    |select * from hadoop_prod.mydb.mytest
  """.stripMargin).show(100)

//10.合并Iceberg表的数据文件
// 1) 首先向表 mytest 中插入一批数据，将数据写入到表mytest中
import spark.implicits._
val df: DataFrame = spark.read.textFile("D:\\2018IDEA_space\\Iceberg-Spark-Flink\\SparkIcebergOperate\\data\\nameinfo")
  .map(line => {
    val arr: Array[String] = line.split(",")
    (arr(0).toInt, arr(1), arr(2).toInt)
  }).toDF("id","name","age")
df.writeTo("hadoop_prod.mydb.mytest").append()

//2) 合并小文件数据,Iceberg合并小文件时并不会删除被合并的文件，Compact是将小文件合并成大文件并创建新的Snapshot。如果要删除文件需要通过Expire Snapshots来实现,targetSizeInBytes 指定合并后的每个文件大小
val conf = new Configuration()
val catalog = new HadoopCatalog(conf,"hdfs://mycluster/sparkoperateiceberg")
val table: Table = catalog.loadTable(TableIdentifier.of("mydb","mytest"))
Actions.forTable(table).rewriteDataFiles().targetSizeInBytes(1024)//1kb，指定生成合并之后文件大小
  .execute()

//11.删除历史快照,历史快照是通过ExpireSnapshot来实现的，设置需要删除多久的历史快照
val conf = new Configuration()
val catalog = new HadoopCatalog(conf,"hdfs://mycluster/sparkoperateiceberg")
val table: Table = catalog.loadTable(TableIdentifier.of("mydb","mytest"))
table.expireSnapshots().expireOlderThan(1640070000000L).commit()

CREATE TABLE ${CataLog名称}.${库名}.${表名} (
  id bigint, 
  name string
) using iceberg
PARTITIONED BY (
  loc string
) TBLPROPERTIES (
    'write.metadata.delete-after-commit.enabled'= true,
	'write.metadata.previous-version-max' = 3
)