我想实现以下案例：
1. select count(1) from user;
2. select count(distinct uid) from user;
但是，聚合操作不能用于非键控流。
非键控流不能使用键控状态，我不知道该怎么办，谁能帮帮我，谢谢！
您可以使用 Table API 执行此操作，但我想知道如何在 DataSteam API 中执行此操作。
【问题讨论】：
标签： apache-flink flink-streaming flink-sql flink-cep
使用 Table API 是推荐的方法。通过使用 DataStream API，您可能会做更多的工作并获得性能较低的解决方案。
但是，回答您的问题：
在这两种情况下，如果您想使用 DataStream API 并且不关心并行执行操作，则可以通过常量对流进行键控，然后使用键控状态。
要并行化count(1) 的情况，您可以通过something 对流进行键控，以便对流进行分区。然后在一组并行计数任务中，您可以对小批量进行计数，并将批量计数更新（作为变更日志流）向下游发送到汇总/报告任务的单个实例。或者您可以使用non-keyed state 构建解决方案。
要并行化 count(distinct user) 案例，您可以根据 Table API 中实现的方法为您的解决方案建模 - 请参阅 the docs。简而言之，它通过将select count(distinct uid) from user 转换为
SELECT SUM(cnt)
FROM (
SELECT COUNT(DISTINCT uid) as cnt
FROM T
GROUP BY MOD(HASH_CODE(uid), 1024)
)
【讨论】：