flink-sql如何处理像“count(Distinct)”这样的场景？

Question

我需要使用flink-sql实时计算“每日活跃用户”，这就像是对每日数据的“count(distinct )”操作。

我的问题是，如果userA像预期的那样在凌晨1点登录，并且flink将DAU加1。现在，userA在晚上10点再次登录，flink-sql怎么知道今天早上userA已经被处理了？是否需要对全天的登录日志重复做count(distinct )？如果不是，flink如何处理这个senario？

Answer 1

Distinct在流媒体中是一个非常昂贵的操作。如果您不使用基于时间的窗口(TUMBLE、SLIDE、SESSION)，则运行时必须将所有值永久存储在状态中，因为它需要假设另一条记录可能在将来的任何时候到达。

但是，您可以设置选项table.exec.state.ttl (see here)使这些记录保持状态的时间。在设计具有长时间运行的查询的SQL管道时，这可能是最重要的选项之一，其中操作符输入的值空间不是恒定的。

Answer 2

所谓实时，我假设你是指在连续查询中？

请参阅https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/tuning/streaming_aggregation_optimization.html

默认情况下，无界聚合运算符逐个处理输入记录，即(1)从状态读取累加器，(2)将记录累加/收回到累加器，(3)将累加器写回状态，(4)下一条记录将从(1)再次执行该过程。

累加器不仅保存最终结果，而且还保留足够的数据来生成下一个结果，而无需再次读取所有先前的记录。

我猜在count(distinct)的情况下，它意味着每天在累加器中保留所有唯一的用户。