Flink SQL - выбрать последний

Question

Я хотел бы передать последнюю запись временного окна. Это легко сделать с помощью maxBy в обычном Flink, но я не могу заставить его работать через SQL API. Я хочу:

SELECT LAST(attribute) FROM [table]
  GROUP BY key, TUMBLE(ts, INTERVAL '1' DAY)

, который ведет себя аналогично

ds.keyBy(key)
  .window(TumblingEventTimeWindows.of(Time.days(1)))
  .maxBy(x -> x.getTs())

Любым способом добиться этого в SQL API?

Answer 1

Я не думаю, что во Flink есть встроенная функция для этого, но вы можете реализовать для этого определяемую пользователем агрегатную функцию.

Вам нужно немного настроить запрос и передать поле отметки времени в функции агрегирования, поскольку SQL не предполагает порядок строк в группе GROUP BY:

SELECT last_by(attribute, ts) FROM [table]
  GROUP BY key, TUMBLE(ts, INTERVAL '1' DAY)

Подробности см. в документации как реализовать и регистр определяемая пользователем функция агрегирования.