Время разворачивания времени обработки вычисляет несогласованные результаты

Question

В Flink я читаю файл с использованием readTextFile и применяю SlidingProcessingTimeWindows.of(Time.milliseconds(60), Time.milliseconds(60)) из 60 мс с слайдом 60 мс на нем. В оконном потоке я вычисляю среднее значение второго поданного кортежа. Мой текстовый файл содержит 1100 строк, а каждая строка - кортеж (String, Integer). Я установил параллелизм в 1 и набрал сообщения в первом поле кортежа.

Когда я запускаю код, каждый раз, когда я получаю разные ответы. Я имею в виду, что кажется, что когда-то он читает весь файл, и когда-то он сначала считывает некоторые строки файла. Имеет ли это отношение к размеру окна скольжения? Как это отношение можно обнаружить, чтобы я мог определить размер и скользящее количество окна?

Answer 1

Ответ в комментарии AlpineGizmo правильный. Я добавлю еще несколько деталей.

Флинк выравнивает окна времени до начала эпохи (1970-01-01-00:00:00). Это означает, что оконный оператор с 1-часовым окном запускает новое окно с каждым новым часом (т. Е. 00:00, 01:00, 02:00, ...), а не с первой прибывающей записью.

Время обработки окон оценивается исходя из текущего времени системы. Как сказано выше, это означает, что количество данных, которое может быть обработано, зависит от ресурсов обработки (аппаратного обеспечения, загрузки процессора/ввода-вывода, ...) машины, на которой работает оператор. Поэтому окно времени обработки не может обеспечить надежные и последовательные результаты.

I Ваш случай, оба описанных эффекта могут привести к результатам, которые не соответствуют друг другу. В зависимости от того, когда вы запустили задание, данные будут назначены различным окнам (если первая запись поступит непосредственно перед закрытием первого 60-секундного окна, только этот элемент будет в окне). В зависимости от нагрузки ввода-вывода устройства может потребоваться больше или меньше времени для доступа и чтения файла.

Если вы хотите иметь согласованные результаты, вам нужно использовать событие-время. В этом случае записи обрабатываются на основе времени, которое кодируется в данных, то есть результаты зависят только от данных, а не от внешних эффектов, таких как время начала задания или загрузка машины обработки.