小编Ran*_*Guy的帖子

我有一个 kafka 流，正在加载到 Spark。来自 Kafka 主题的消息具有以下属性：bl_iban、blacklisted、timestamp。因此，有 IBANS、关于该 IBAN 是否被列入黑名单 (Y/N) 的标志，并且还有该记录的时间戳。问题是一个 IBAN 可以有多个记录，因为超时的 IBAN 可能会被列入黑名单或“删除”。我想要实现的目标是了解每个 IBANS 的当前状态。然而，我从更简单的目标开始，那就是列出每个最新的 IBAN timestamp（之后我也想添加blacklisted状态），所以我生成了以下代码（其中黑名单代表我从 Kafka 加载的数据集）：

blackList = blackList.groupBy("bl_iban")
                .agg(col("bl_iban"), max("timestamp"));

之后我尝试使用以下代码将其打印到控制台：

StreamingQuery query = blackList.writeStream()
    .format("console")
    .outputMode(OutputMode.Append())
    .start();

我已经运行我的代码并收到以下错误： Append output mode not supported when there are streaming aggregations on streaming DataFrames/DataSets without watermark

所以我将水印添加到我的数据集中，如下所示：

blackList = blackList.withWatermark("timestamp", "2 seconds")
                .groupBy("bl_iban")
                .agg(col("bl_iban"), max("timestamp"));

之后又出现同样的错误。我有什么想法可以解决这个问题吗？

更新：在迈克的帮助下，我成功地摆脱了这个错误。但问题是我仍然无法让我的黑名单发挥作用。我可以看到数据是如何从 Kafka 加载的，但之后从我的组操作中我得到了两个空批次，仅此而已。从Kafka打印的数据：

+-----------------------+-----------+-----------------------+
|bl_iban                |blacklisted|timestamp              |
+-----------------------+-----------+-----------------------+
|SK047047595122709025789|N …