Блог

В настоящее время я использую FsStateBackend для проверки состояния. Я использую интервал 10 секунд, как показано ниже. Но я вижу, что стоимость пакета передачи, который использует checkpoint, составляет примерно 20 долларов в день, а стоимость aws transfer s3: 0,005 доллара за 1000 запросов => (я использую ~ 4000000 запросов / день @@). У меня есть 7 заданий, которые:

• 6 заданий с использованием интервала контрольных точек = 10000 (мс)
• 1 задание с использованием интервала контрольных точек = 1000 (мс)

И запустите flink на AWS EMR. Средний размер состояния для каждой контрольной точки от (8 КБ -> 30 М). Что произошло за контрольной точкой?

 // set up checkpoint
        env.enableCheckpointing(1000 or 10000);

        // advanced options:
        // make sure 500 ms of progress happen between checkpoints
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        // checkpoints have to complete within one minute, or are discarded
//            env.getCheckpointConfig().setCheckpointTimeout(60000);
        // allow only one checkpoint to be in progress at the same time
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // enable externalized checkpoints which are retained after job cancellation
        env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // folder to checkpoint
        StateBackend backend = new FsStateBackend(checkpointPath, true);
        env.setStateBackend(backend);
  

Какую реализацию S3 вы используете для контрольной точки? Это имеет большое значение.

Хотя вы должны использовать реализацию hadoop для S3 с StreamingFileSink , это может быть плохим выбором для контрольной точки. Hadoop S3 FS пытается имитировать файловую систему поверх S3:

• перед записью ключа он проверяет, существует ли «родительский каталог», проверяя наличие ключа с префиксом до последнего «/»
• он создает пустые файлы-маркеры, чтобы отметить существование такого родительского каталога
• все эти запросы «существования» являются дорогостоящими запросами S3 HEAD

В результате у Hadoop S3 FS очень высокая задержка «создания файла», и он быстро сталкивается с ограничениями скорости запросов (у запросов HEAD очень низкие ограничения скорости запросов на S3).

Presto S3 не пытается творить эту магию; он просто выполняет операции PUT / GET без всего остального. Поскольку контрольная точка Flink не предполагает ничего большего, она более эффективна и последовательна.

Кроме того, с Hadoop S3 вы можете столкнуться с ситуацией, когда вы не выполняете операции восстановления, потому что похоже, что файла состояния там нет (запрос HEAD приводит к ложному кэшированию в балансировщике нагрузки S3). Только через некоторое время файл будет виден, и только тогда восстановление будет успешным.

Однако обратите внимание, что существуют также проблемы с использованием Presto S3 для контрольных точек. См. FLINK-24392. Ни одна из реализаций не идеальна.

Можно, кстати, использовать как версию hadoop для приемника, так и версию presto для контрольных точек. В этом случае вы должны явно использовать s3a:// в качестве схемы для приемника (Hadoop) и s3p:// для контрольной точки (Вуаля).

Документы Flink S3.