如果您对引用“带水印的时间”（即1小时）感兴趣，您可以替换 watermark 在表达式中 current_timestamp - interval '1' hour .
由于您试图连接两个流，spark将坚持两者都使用水印

参考

spark流到流连接

对于内部流连接，水印和事件时间约束（连接条件）是可选的。
如果一个无界状态在卷方面不是您的问题，您可以选择不指定它们。在这种情况下，所有数据都将被缓冲，来自faststream的数据将立即与来自slowstream的所有数据连接起来。
只有当两个参数都被指定时，您的状态才会被清除。请注意这两个参数的用途：
事件时间约束（时间范围连接条件）：两个事件在各自的源上生成之间的最大时间范围是多少？
水印：事件在源和处理引擎之间传输时可以延迟的最长持续时间是多少？
要定义这两个参数，首先需要回答上述问题（引自o`reilly出版的“learning apache spark，2nd edition”一书）。
关于您的代码注解：
“通过‘告诉’spark永远不会有新的匹配项，防止联接缓冲快速流。”
记住，流连接中的缓冲是必要的。否则，您只能加入当前微批处理中可用的数据。由于slowstream没有定期更新，但是faststream更新数据的速度非常快，如果不缓冲数据，您可能永远不会得到任何连接匹配。
总的来说，对于您描述的用例（“将快速变化的数据与缓慢变化的元数据连接起来”），通常最好使用流静态连接方法，其中缓慢变化的数据成为静态部分。
在流静态联接中，流数据中的每一行都将与完整的静态数据联接，而静态表则加载在每个微批中。如果加载静态表会降低性能，您可以考虑缓存它并定期更新它，如stream static join:how to refresh（unpersist/persist）static dataframe periodically中所述。

用我最终得到的回答我自己的问题。这当然不理想，但对于我所有的搜索来说，spark structured streaming中似乎没有控件来处理这个用例。
所以我的解决方案是读取数据集并在 foreachBatch . 这样我可以防止spark存储大量不必要的状态，并立即执行连接。缺点是，似乎没有办法增量读取流表，所以每次我都会重新读取整个表。。。

def join_slow_stream(df, batchID):

  # Read as a table rather than a stream
  slowdf = spark.read \
    .format("delta") \
    .load("dbfs:/mnt/some_file/slowStream") \
    .alias("slowStream")

  out_df = df.join(
    slowdf,
    expr(""" 
      fastStream.slow_id = slowStream.id
      """
    ),
    "inner"
  ).select("fastStream.*", "slowStream.metadata")

  # write data to database
  db_con.write(out_df)

fastStream.writeStream.foreachBatch(join_slow_stream)