I'm trying to do the below execution in Spark 3.1.2 / Scala 2.12:
streamDF.createOrReplaceTempView("tempStream")
val query = spark.sql(
s"""
|MERGE INTO catalog.db.tablename t
|USING tempStream s
|ON t.ID = s.ID
| WHEN MATCHED THEN UPDATE SET *
| WHEN NOT MATCHED THEN INSERT *
|""".stripMargin
)
query
.writeStream
.format("console")
.outputMode(OutputMode.Update)
.start
.awaitTermination
But I'm getting this exception:
Queries with streaming sources must be executed with writeStream.start()
However, the below execution works well:
streamDF.createOrReplaceTempView("tempStream")
val query = spark.sql(
s"""
|SELECT COUNT(*) FROM tempStream
|""".stripMargin
)
query
.writeStream
.format("console")
.outputMode(OutputMode.Update)
.start
.awaitTermination
Can someone please explain why is this happening? Is there some alternative for MERGE INTO in this case?