有没有一个很好的方法来加入spark中的流和变更表？

val spark = SparkSession.builder()
  .appName("Streaming and Delta Lake")
  .master("local[*]")
  .getOrCreate()

import org.apache.spark.sql.streaming.Trigger
import io.delta.tables._

val streamDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "topic")
  .load()

streamDF.createOrReplaceTempView("stream_view")

val deltaTable = DeltaTable.forPath(spark, "/path/to/delta_table")

deltaTable.toDF.createOrReplaceTempView("delta_table_view")

val resultDF = spark.sql("""
  SELECT *
  FROM stream_view
  JOIN delta_table_view
  ON stream_view.key = delta_table_view.key
""")

resultDF.writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("topic", "output_topic")
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .start()
  .awaitTermination()