Spark структурированная потоковая агрегация в реальном времени

Question

Можно ли выводить данные агрегации по каждому триггеру до истечения временного окна агрегации?

Контекст: Я разрабатываю приложение, которое читает данные из топологии Кафки c, обрабатывают данные, агрегируют их за 1 час и выводят на S3. Однако приложение Spark вполне понятно выводит данные агрегации на S3 только в конце данного часового окна.

Проблема заключается в том, что конечные пользователи агрегированных данных на S3 могут иметь только половину реального времени. посмотрим, так как они всегда на час опаздывают, ожидая следующей агрегации, которая будет выведена из приложения spark.

Сокращение временного окна агрегации до значения, меньшего чем час, безусловно, поможет, но будет генерировать намного больше data.

Что можно сделать, чтобы включить агрегацию в реальном времени, как я ее называю, с использованием минимальных ресурсов?

Answer 1

Это интересное предложение, и у меня есть предложение, но я не уверен, что оно действительно соответствует вашим минимальным критериям. В любом случае, я опишу решение ...

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

Идея проста - просто вводите данные в первую базу данных, а затем продолжайте выгружать данные в S3 через определенное время, т.е. через час или день, в зависимости от ваших требований. Затем вы можете зарегистрировать метаданные обоих этих уровней хранения в слое метаданных (например, AWS Glue) - это может быть не всегда необходимым, если вам не нужно постоянное метасторство. Кроме того, вы можете использовать что-то вроде Presto для запроса в обоих этих магазинах. Это также позволит вам оптимизировать хранилище в этих двух хранилищах данных.

Вам, очевидно, потребуется создать процесс для удаления / удаления разделов данных из хранилища, в которое вы будете передавать данные, а также для перемещения данные в S3.

Эта модель называется многоуровневой моделью хранения или иерархической моделью хранения с шаблоном скользящего окна - Справочная статья от Cloudera .

Надеюсь, это поможет!