nareshbab commented on issue #13431:
URL: https://github.com/apache/iceberg/issues/13431#issuecomment-3021744943
@RussellSpitzer Here's the rudimentary code to replicate this. I hope this
provides the clarity on the process
Overall steps executed:
- Start streaming pipeline
- post a message in kafka topic with schema as shared below
- Batch N Starts
- Write to Audit table
- Merge to snapshot table
- Batch N ends
- Run another spark script with iceberg configurations to same db
- query from spark script "select * from snapshot" & "select * from
temp_audit_table" to see current rows reflected in the shell. Data is visible
in audit table but not in snapshot table
- Send another message in kafka whenever user desires
- Batch N+1 starts
- Write to Audit table
- Merge to snapshot table
- Batch N ends
- Run another spark script to query data "select * from snapshot" & "select
* from temp_audit_table" to see current rows reflected in the shell. Data from
Batch N is visible in snapshot. But data from Batch N+1 is only visible in
audit table
`
// Required imports
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.{Dataset, Row, SaveMode, SparkSession}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.functions.{col, current_date, from_json}
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType,
StructField, StructType, TimestampType}
object MergeTest extends Logging {
// Configuration variables
val catalogName = "local"
val dbName = "biglake_db_us_central1
[batchN_logs.txt](https://github.com/user-attachments/files/20990963/batchN_logs.txt)
[batchN+1_logs.txt](https://github.com/user-attachments/files/20990962/batchN%2B1_logs.txt)
"
val tableName = "snapshot_table"
val tempTableName = "temp_audit_table"
val warehouse = "<>/data/warehouse"
val checkpointLocation = "<>/merge-test-checkpoint"
def main(args: Array[String]): Unit = {
// Spark session with Iceberg Local catalog configuration
implicit val spark = SparkSession.builder()
.appName("Iceberg to BigLake")
.master("local[1]")
.config("spark.sql.extensions",
"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
.config(s"spark.sql.catalog.${catalogName}",
"org.apache.iceberg.spark.SparkCatalog")
.config(s"spark.sql.catalog.${catalogName}.type", "hadoop")
.config(s"spark.sql.catalog.${catalogName}.warehouse", warehouse)
.getOrCreate()
val kafkaDF = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("subscribe", "user-updates")
.option("startingOffsets", "earliest")
.option("maxOffsetsPerTrigger", 1000)
.option("kafka.security.protocol", "PLAINTEXT")
.load()
val schema = StructType(Array(
StructField("record_id", StringType, nullable = false),
StructField("field1", StringType, nullable = true),
StructField("field2", IntegerType, nullable = true),
StructField("field3", DoubleType, nullable = true),
StructField("operation", StringType, nullable = false), // INSERT,
UPDATE, DELETE
StructField("event_timestamp", TimestampType, nullable = false)
))
val enrichedDF = kafkaDF
.select(
from_json(col("value").cast("string"), schema).as("update_data"),
current_date().as("partition_date")
)
.select(
col("update_data.*"),
col("partition_date")
)
.filter(col("record_id").isNotNull)
enrichedDF
.writeStream
.foreachBatch((batchDF: Dataset[Row], batchId: Long) => {
writeAuditTable(batchDF, batchId)
executeMerge(batchDF, batchId)
})
.option("checkpointLocation", checkpointLocation)
.trigger(Trigger.ProcessingTime("30 seconds"))
.queryName("KafkaToSnapshotPipeline")
.start()
.awaitTermination()
}
def writeAuditTable(batchDF: Dataset[Row], batchID: Long)(implicit spark:
SparkSession): Unit = {
logInfo(s"Writing audit table for batch $batchID")
ensureTableExists()
try {
// Write the batch to the temp audit table
batchDF
.writeTo(s"$catalogName.$dbName.$tempTableName")
.replace()
logInfo(s"Successfully wrote audit table for batch $batchID")
} catch {
case e: Exception =>
logError(s"Failed to write audit table for batch $batchID", e)
throw e
}
}
def executeMerge(batchDF: Dataset[Row], batchId: Long)(implicit spark:
SparkSession): Unit = {
logInfo(s"Executing Iceberg merge operation for batch $batchId")
ensureTableExists()
try {
// Execute Iceberg MERGE INTO statement
val mergeSQL = s"""
MERGE INTO $catalogName.$dbName.$tableName AS target
USING $catalogName.$dbName.$tempTableName AS source
ON target.record_id = source.record_id
WHEN MATCHED AND source.operation = 'DELETE' THEN
DELETE
WHEN MATCHED AND source.operation IN ('UPDATE', 'INSERT') THEN
UPDATE SET *
WHEN NOT MATCHED AND source.operation IN ('INSERT', 'UPDATE') THEN
INSERT *
"""
logInfo(s"Executing Iceberg MERGE: ${mergeSQL.take(600)}...")
spark.sql(mergeSQL)
logInfo(s"Successfully executed Iceberg merge for batch $batchId")
} catch {
case e: Exception =>
logError(s"Failed to execute Iceberg merge for batch $batchId", e)
throw e
}
}
def ensureTableExists()(implicit spark: SparkSession): Unit = {
val tableExists =
spark.catalog.tableExists(s"$catalogName.$dbName.$tableName")
val tempTableExists =
spark.catalog.tableExists(s"$catalogName.$dbName.$tempTableName")
if (!tableExists) {
logInfo(s"Table $catalogName.$dbName.$tableName does not exist,
creating it")
val createTableSQL = s"""
CREATE TABLE IF NOT EXISTS $catalogName.$dbName.$tableName (
record_id STRING,
field1 STRING,
field2 INT,
field3 DOUBLE,
operation STRING,
event_timestamp TIMESTAMP,
partition_date DATE
)
USING iceberg
PARTITIONED BY (partition_date)
"""
spark.sql(createTableSQL)
logInfo(s"Created Iceberg Table $catalogName.$dbName.$tableName")
} else {
logInfo(s"Table $catalogName.$dbName.$tableName already exists")
}
//Creating temp table
val createTempTableSQL = s"""
CREATE TABLE $catalogName.$dbName.$tempTableName (
record_id STRING,
field1 STRING,
field2 INT,
field3 DOUBLE,
operation STRING,
event_timestamp TIMESTAMP,
partition_date DATE
)
USING iceberg
"""
if (!tempTableExists) {
logInfo(s"Temp table $catalogName.$dbName.$tempTableName does not
exist, creating it")
spark.sql(createTempTableSQL)
logInfo(s"Created Temp Iceberg Table
$catalogName.$dbName.$tempTableName")
}
}
}
`
Kafka Message
`
{
"record_id": "user_001",
"field1": "Carol Davis",
"field2": 25,
"field3": 88.7,
"operation": "INSERT",
"event_timestamp": "2024-06-24T09:02:00.000Z"
}
`
Attaching below are the logs for batch N and batch N+1 from another spark
script to check table state between batches
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
