kickthetires's Introduction

Universal Format (UniForm) for Iceberg compatibility with Delta tables

Config Unity Catalog
Create a Catalog
Use the Catalog and create a table with TBLPROPERTIES

import org.apache.spark.sql.types._
val ucName = "uc_hsshin"
val schemaName = "kickthetires"
val icebergTableName = "clickEvents"   
val streamingDataFrameSchema: StructType = flattened_df.schema // Replace result_df with your streaming DataFrame  

spark.sql(s"DROP TABLE IF EXISTS $ucName.$schemaName.$icebergTableName")
spark.sql(s"DROP DATABASE IF EXISTS $ucName.$schemaName")

// Create Database 
spark.sql(s"CREATE DATABASE $ucName.$schemaName")
spark.catalog.setCurrentDatabase(schemaName) // Can't find the database

// Create an empty Delta table with the same schema as your streaming DataFrame  
spark.sql(s"""  
  CREATE TABLE IF NOT EXISTS $ucName.$schemaName.$icebergTableName (  
    ${streamingDataFrameSchema.toDDL}  
  )  
  TBLPROPERTIES(
  'delta.universalFormat.enabledFormats' = 'iceberg');
""")

Start to write stream to the table

import org.apache.spark.sql.streaming.Trigger  
  
val streamQuery = flattened_df.writeStream  
  .format("iceberg")  
  .outputMode("append")
  .option("checkpointLocation", icebergcheckpointLocation) // Choose an appropriate checkpoint location  
  .trigger(Trigger.ProcessingTime("30 seconds")) // Data will be written every 30 seconds
  .table(s"$ucName.$schemaName.$icebergTableName") // Write to the iceberg table  
  
streamQuery.awaitTermination()