Flink SQL Continuous Window Top-N 查询结果不一致答案

【问题标题】：Inconsistent results for Flink SQL Continuous Window Top-N queryFlink SQL Continuous Window Top-N 查询结果不一致
【发布时间】：2021-11-04 11:38:08
【问题描述】：

对于 Flik Kafka SQL 源定义：

CREATE TABLE PlayEvents (
  `event_time` TIMESTAMP(3) METADATA FROM 'timestamp',
  `the_kafka_key` STRING, 
  `song_id` BIGINT NOT NULL, 
  `duration` BIGINT, 
  WATERMARK FOR `event_time` AS `event_time` - INTERVAL '1' SECONDS 
) WITH (
  'connector' = 'kafka', 
  'topic' = 'play-events',
  'properties.bootstrap.servers' = 'localhost:29092',
  'key.format' = 'raw',
  'key.fields' = 'the_kafka_key',
  'value.format' = 'avro-confluent',
  'value.avro-confluent.url' = 'http://localhost:8081',
  'value.fields-include' = 'EXCEPT_KEY',
  'scan.startup.mode' = 'earliest-offset'
)

测试PlayEvent 消息每100ms 发送到play-events 使用 Avro 格式：

protocol `protocol` {
    record PlayEvent {
        long song_id;
        long duration;
    }
}

我正在运行一个连续的Window Top-N 查询：

Configuration strConf = new Configuration();
strConf.setInteger(RestOptions.PORT, 8089);
strConf.setString(RestOptions.BIND_PORT, "8088-8090");

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(strConf);
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
Configuration tableConfiguration = tableEnv.getConfig().getConfiguration();
tableConfiguration.setString("table.exec.source.idle-timeout", "2 min");

tableEnv.sqlQuery("" +
    "SELECT window_end, song_id, play_count FROM ( " +
    " SELECT *, " +
    "    ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY play_count DESC) AS row_num " +
    " FROM (" +
    "    SELECT window_start, window_end, song_id, COUNT(*) AS play_count " +
    "    FROM TABLE( " +
    "      TUMBLE(TABLE PlayEvents, DESCRIPTOR(event_time), INTERVAL '60' SECONDS)) " +
    "    GROUP BY window_start, window_end, song_id " +
    " ) " +
    ") WHERE row_num <= 4 "
).execute().print();

第一个打印输出在开始后显示一致的结果，每个窗口期恰好有 4 个结果：

...
| +I | 2021-11-04 12:23:00.000 |                    3 |                   16 |
| +I | 2021-11-04 12:23:00.000 |                    6 |                   14 |
| +I | 2021-11-04 12:23:00.000 |                    1 |                   12 |
| +I | 2021-11-04 12:23:00.000 |                   11 |                   10 |

| +I | 2021-11-04 12:24:00.000 |                    9 |                   16 |
| +I | 2021-11-04 12:24:00.000 |                    6 |                   13 |
| +I | 2021-11-04 12:24:00.000 |                    7 |                   12 |
| +I | 2021-11-04 12:24:00.000 |                    5 |                   12 |

后续输出虽然显示了部分窗口内容：

...
| +I | 2021-11-04 12:25:00.000 |                    9 |                   12 |
| +I | 2021-11-04 12:25:00.000 |                    6 |                    6 |

| +I | 2021-11-04 12:27:00.000 |                   11 |                   18 |
...

例如该窗口每个区间仅包含 1、2、3 个元素，有时整个区间都会被删除。

如果我重新启动查询，所有结果将在第一次打印时正确显示（例如，每个窗口恰好 4 个元素..），然后它将继续部分响应。

我尝试了不同的配置，尝试将连续查询写入另一个 Kafka 主题，然后从那里查询。但不一致的部分结果仍然存在。

不确定我是否遗漏了一些重要配置，或者这是 Flink 的已知限制或问题？

附：我正在使用 Flink 1.14 在本地机器上运行我的测试进行测试。

【问题讨论】：

标签： flink-streaming flink-sql

【解决方案1】：

当执行事件时间处理的 Flink 作业表现出不确定的行为时，这是因为处理时间在某种程度上会影响结果。在这种情况下，我的猜测是实际的乱序超过了水印中配置的 1 秒间隔，导致延迟事件被丢弃。

idle-timeout 也可以在这里发挥作用。当标记为空闲的源恢复传递事件时，很容易出现水印已经超过来自先前空闲源的初始事件中的时间戳的情况。即使一秒钟足以处理正常预期的乱序，在这种特殊情况下也可能不够长。

【讨论】：

谢谢大卫！我一直在尝试不同的水印间隔，但没有成功。为了实验，我刚刚将 PlayEvents 的 Watermark 增加到 30 秒，并将空闲超时时间增加到 5 分钟。不幸的是还是一样的结果。我求助于空闲超时，因为没有它，查询永远不会开始。所以在这种情况下，我必须等待 5 分钟才能弹出第一个结果。然后它（或多或少）按预期每 1 m 打印一次结果。只是结果不完整。
会不会有没有 4 首不同歌曲数据的区间？此外，您必须设置空闲超时是不寻常的。这表明您至少有一个没有太多数据的 Kafka 分区。
...没有 4 首不同歌曲的数据？不太可能。从 14 首歌曲中发送 10 msg/sec。此外，如果我重新运行相同的查询，第一次打印将正确显示所有结果（例如 4 个条目/窗口），即使对于它在上次运行中确实错过的窗口也是如此。 您必须设置空闲超时是不寻常的。这表明您至少有一个没有太多数据的 Kafka 分区。 有没有办法验证这一点？这是我用于测试的 Kafka+SchemaReg：docker-compose-kafka.yml
还有，这不是暗示play-events主题只有一个分区：Topic: play-events TopicId: KEzc4NxISqySFa6czLKMuw PartitionCount: 1 ReplicationFactor: 1 Configs: Topic: play-events Partition: 0 Leader: 1001 Replicas: 1001 Isr: 1001

【解决方案2】：

问题似乎与等于机器核心数的defaultLocalParallelism （在我的情况下为16）有关。结果产生了 16 个子任务but only one is used 来处理输入数据（输入的 Kafka 主题仅使用 1 个分区）。

如果我将parallelism 显式设置为1，则空闲超时并解决查询结果不一致问题：

Configuration strConf = ...
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1, strConf);

或者你可以这样设置：

env.getConfig().setParallelism(1);

附：认为这解决了我的演示问题，我想知道为本地环境配置并行度/最大并行度的正确方法是什么。

【讨论】：