apache-spark - Spark Structured Streaming foreachBatch 和 UPSERT(合并): to persist or not to persist?

Question

如果在使用 foreachBatch 的结构化流中进行状态聚合(任意)以将更新合并到增量表中，我是否应该在更新之前将批处理数据帧保留在 foreachBatch 中？

似乎不需要持久化，因为我正在写入单个数据接收器。

另一方面，我有强烈的感觉，不坚持会导致源重新扫描并触发聚合两次。

有什么意见/想法吗？

foreachBatch((VoidFunction2<Dataset<Row>, Long>) (batchDf, batchId) -> 
        deltaTable.as("table").merge(batchDf.as("updates"), functions.expr("table.id=updates.id"))
        .whenNotMatched().insertAll()           // new session to be added
        .whenMatched()
        .updateAll()
        .execute())

Answer 1

因此 delta 用户 (https://groups.google.com/g/delta-users/c/Ihm6PMilCdI) 的答案是:

DeltaTable.merge (upsert) does two passes on the source data.

因此，如果您确实关心 mapGroupsWithState 内任意状态聚合中的 Spark 指标或日志/flatmapGroupsWithState - 在内部合并之前进行持久化/缓存 foreachBatch , 否则发送的指标将具有双倍 (x2) 值并且日志聚合日志将被发出两次