我已经为另一个问题开发了一个解决方案Stream-Static Join: How to refresh (unpersist/persist) static Dataframe periodically,这个解决方案可能也有助于解决你的问题:
你可以利用Structured Streaming提供的流调度功能来触发静态Dataframe的刷新(取消持久化 -> 加载 -> 持久化)。具体步骤如下:
- 首先加载静态Dataframe并保存为
var
- 定义一个刷新静态Dataframe的方法
- 使用一个"Rate"流,在所需的时间间隔(例如1小时)触发刷新操作
- 读取实际的流数据,并与静态Dataframe进行连接操作
- 在Rate流中使用
foreachBatch
接收器调用刷新方法
以下代码在Spark 3.0.1、Scala 2.12.10和Delta 0.7.0上运行良好。
var staticDf = spark.read.format("delta").load(deltaPath)
staticDf.persist()
def foreachBatchMethod[T](batchDf: Dataset[T], batchId: Long) = {
staticDf.unpersist()
staticDf = spark.read.format("delta").load(deltaPath)
staticDf.persist()
println(s"${Calendar.getInstance().getTime}: Refreshing static Dataframe from DeltaLake")
}
val staticRefreshStream = spark.readStream
.format("rate")
.option("rowsPerSecond", 1)
.option("numPartitions", 1)
.load()
.selectExpr("CAST(value as LONG) as trigger")
.as[Long]
val streamingDf = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("subscribe", "test")
.option("startingOffsets", "earliest")
.option("failOnDataLoss", "false")
.load()
.selectExpr("CAST(value AS STRING) as id", "offset as streamingField")
val joinDf = streamingDf.join(staticDf, "id")
val query = joinDf.writeStream
.format("console")
.option("truncate", false)
.option("checkpointLocation", "/path/to/sparkCheckpoint")
.start()
staticRefreshStream.writeStream
.outputMode("append")
.foreachBatch(foreachBatchMethod[Long] _)
.queryName("RefreshStream")
.trigger(Trigger.ProcessingTime("5 seconds"))
.start()
为了有一个完整的例子,Delta表应该按照下面的方式创建:
val deltaPath = "file:///tmp/delta/table"
import spark.implicits._
val df = Seq(
(1L, "static1"),
(2L, "static2")
).toDF("id", "deltaField")
df.write
.mode(SaveMode.Overwrite)
.format("delta")
.save(deltaPath)
mapGroupsWithState
来处理刷新静态数据集的自定义逻辑。 - Paul LeclercqaddBatch
方法中刷新我的数据集。希望这个例子能够帮到你:https://github.com/polomarcus/Spark-Structured-Streaming-Examples/blob/master/src/main/scala/cassandra/StreamSinkProvider/CassandraSink.scala#L39 - Paul Leclercq