如何在不使用 flatMapsGroupWithState 的情况下使用 Structured Streaming 2.3.0 在 spark 中进行无状态聚合？

Question

如何在不使用 flatMapsGroupWithState 或 Dstream API 的情况下使用 Structured Streaming 2.3.0 在 spark 中进行无状态聚合？寻找一种更具声明性的方式

例子：

select count(*) from some_view

我希望输出只计算每批中可用的任何记录，但不计算前一批的汇总

Answer 1

要在 spark 中使用 Structured Streaming 2.3.0 而不使用 flatMapsGroupWithState 或 Dstream API 进行无状态聚合，您可以使用以下代码-

import spark.implicits._

def countValues = (_: String, it: Iterator[(String, String)]) => it.length

val query =
  dataStream
    .select(lit("a").as("newKey"), col("value"))
    .as[(String, String)]
    .groupByKey { case(newKey, _) => newKey }
    .mapGroups[Int](countValues)
    .writeStream
    .format("console")
    .start()

我们正在做的是-

1. 我们在datastream - newKey 中添加了一列。我们这样做是为了可以使用groupByKey 对其进行groupBy。我使用了文字字符串"a"，但你可以使用任何东西。此外，您需要从datastream 的可用列中选择任何列。我为此选择了value 列，您可以选择任何人。
2. 我们创建了一个映射函数 - countValues，通过编写 it.length 来计算 groupByKey 函数聚合的值。

因此，通过这种方式，我们可以计算每批中可用的任何记录，但不是从前一批中汇总的记录。

希望对你有帮助！