Flink Autojoin со столбцом времени строки

Ваш запрос не работает по нескольким причинам.

SELECT 
  Id1, MAX(myTimestamp) as myTimestamp, SUM(value) as value 
FROM 
  MyTable, 
  (SELECT Id2, MAX(myTimestamp) as latestTimestamp 
   FROM MyTable 
   GROUP BY Id2
  ) as RecordsLatest
WHERE 
  MyTable.Id2 = RecordsLatest.Id2 
  AND MyTable.myTimestamp = RecordsLatest.myTimestamp
GROUP BY 
  Id1, HOP(rowtime, INTERVAL 10 SECOND, INTERVAL 30 SECOND)

Некоторые из них связаны с ограничениями Flink, другие - более фундаментальными.

1. latestTimestamp больше не является атрибутом rowtime. Это потому, что это вычислено. Как только вы используете атрибут rowtime в выражении (включая функции агрегирования, такие как MAX), они теряют свое свойство rowtime и становятся обычными атрибутами TIMESTAMP.
2. Внутренний запрос создает динамическую таблицу, которая обновляет свои результаты. Это не таблица, предназначенная только для добавления. Как только максимальная временная метка Id2 изменится, предыдущую строку результатов необходимо отозвать и вставить новую строку результатов.
3. Поскольку RecordsLatest является обновляемой таблицей (а не таблицей только для добавления), а latestTimestamp не является атрибутом rowtime, соединение RecordsLatest и MyTable является "обычным соединением" (а не соединением с временным окном), которое также дает результат обновления. и не результат только добавления. Обычное соединение не может создавать какие-либо атрибуты rowtime, потому что нет гарантии относительно порядка вывода строк (что является предварительным условием для атрибутов rowtime, потому что они должны быть выровнены с водяными знаками), и в будущем может потребоваться удалить результаты. Это вызывает сообщение об ошибке, которое вы видите.
4. Предложение GROUP BY внешнего запроса требует входной таблицы только для добавления с атрибутом rowtime rowtime. Однако выходные данные соединения предназначены не только для добавления, но и для обновления, и атрибут rowtime не может быть атрибутом rowtime, как объяснялось ранее.

К сожалению, решить вашу задачу непросто, но вполне возможно.

Прежде всего, вы должны вернуть запрос, который возвращает для каждого (Id1, Id2) окна значения для строки с максимальной временной меткой:

SELECT 
  Id1, Id2,
  MAX(myTimestamp) AS maxT
  ValOfMaxT(valX, myTimestamp) AS valXOfMaxT,
  HOP_ROWTIME(myTimestamp, INTERVAL '10' SECOND, INTERVAL '30' SECOND) AS rowtime
FROM
  MyTable
GROUP BY
  Id1, Id2, HOP(myTimestamp, INTERVAL '10' SECOND, INTERVAL '30' SECOND)

Функция ValOfMaxT - это определяемая пользователем функция агрегирования, которая определяет значение максимальной отметки времени и возвращает его. rowtime - это новый атрибут rowtime и за 1 мс до конечной отметки времени окна.

Учитывая эту таблицу, назовем ее Temp, вы можете определить следующий запрос как:

SELECT
  Id1, MAX(maxT) as myTimestamp, SUM(valXOfMaxT)
FROM Temp
GROUP BY
  Id1, TUMBLE(rowtime, INTERVAL '10' SECONDS)

Этот запрос группирует только Id1 и TUMBLE окно. Это TUMBLE окно, потому что первое HOP окно уже сгруппировало каждую запись в три окна, и мы не должны делать это снова. Вместо этого мы группируем результат первого запроса в 10 вторых окон, потому что это длина слайда HOP окон в первом запросе.