Как устроен механизм отказоустойчивого выполнения в Trino, чем политика повтора QUERY отличается от TASK, зачем настраивать диспетчер обмена на внешнее S3-совместимое хранилище и задавать коэффициент задержки перед повторными попытками выполнить SQL-запрос.
2 политики отказоустойчивого выполнения в Trino
Будучи движком online-обработки больших объемов данных с помощью распределенных SQL-запросов, Trino должен иметь высокую отказоустойчивость, чтобы справляться с такими нагрузками и быстро восстанавливаться в случае внезапного сбоя, например, из-за OOM-ошибки. Для этого в Trino есть механизм отказоустойчивого выполнения, который позволяет кластеру смягчать сбои запросов, повторяя их целиком или частично. При включенном отказоустойчивом выполнении промежуточные данные обмена помещаются в буфер и могут быть повторно использованы другим рабочим процессом в случае сбоя текущего или другой ошибки во время выполнения запроса.
Одна из самых частых причин сбоя – недостаток ресурсов на узле для выполнения задачи. Чем больше время выполнения запроса, тем больше вероятность сбоя. Важно отметить, что сбоем не считается некорректный SQL-запрос или ошибка пользователя: в таких случаях Trino не тратит ресурсы на повторные попытки выполнения неверно поставленной задачи.
По умолчанию отказоустойчивое выполнение отключено. Чтобы включить эту функцию, надо установить свойство конфигурации retry-policy в значение QUERY или TASK в зависимости от желаемой политики повтора. Если retry-policy равно QUERY, вся операция запроса будет автоматически повторяться в случае сбоя. Это полезно при временном сбое, когда следующая попытка выполнения всего запроса с большей вероятностью будет успешной. Такой простой подход гарантирует целостность выполнения всего запроса. Но его транзакционный характер предполагает атомарность, т.е. перезапуск целиком. Это может требовать много времени и ресурсов для больших и сложных запросов. Кроме того, возможны повторные действия, если часть запроса уже была выполнена до сбоя. Поэтому политика повтора QUERY рекомендуется, когда большая часть рабочей нагрузки кластера Trino состоит из множества небольших запросов. По умолчанию Trino не реализует отказоустойчивость для запросов, у которых размер возвращаемых результатов более 32 МБ. Этот предел можно увеличить, изменив свойство конфигурации exchange.deduplication-buffer-size, увеличив его значение по умолчанию, равное 32MB. Но это приведет к более высокому потреблению памяти на координаторе.
При установке retry-policy в значение TASK будут повторно запускаться только неуспешные задачи внутри запроса, а остальные продолжат выполняться. Это эффективно при выполнении распределенных запросов, где отдельные задачи могут столкнуться с временными проблемами, такими как сетевые сбои или перегрузка отдельных узлов. Эта политика рекомендуется при выполнении больших пакетных запросов. При использовании этой политик рекомендуется установить свойство управления запросами task.low-memory-killer.policy в значение total-reservation-on-blocked-nodes, чтобы избежать ручного завершения запросов, если кластеру не хватит памяти. Поэтому при использовании TASK-политики повторных попыток важно управлять объемом данных, обрабатываемых в каждой задаче. Если задачи слишком малы, их координация может занять больше времени обработки и ресурсов, чем выполнение самих задач. Если задачи слишком велики, то для одной задачи может требоваться больше ресурсов, чем доступно на узле, поэтому запросы не выполнятся. Trino поддерживает ограниченное автоматическое изменение размера задачи, о чем мы поговорим в другой раз. Политика повтора TASK лучше подходит для больших пакетных запросов, а для коротких и частых она может привести к задержке. Поэтому лучше всего разделять кластеры Trino с разными политиками повтора: TASK для больших пакетных запросов и QUERY для небольших, но частых. По сравнению с политикой QUERY, в TASK время восстановления после сбоя будет меньше, так как повторяется только часть запроса. Однако, с технической точки зрения это сложнее, особенно при stateful-нагрузках, когда необходимо управлять состоянием выполнения задач. Это может потребовать более сложной логики обработки зависимостей между задачами внутри запроса. Если необходимо переключиться с одного режима на другой в кластере Trino, сперва надо деактивировать политику повтора, установив ее в значение NONE.
Настройки отказоустойчивого выполнения
Примечательно, что включение политики повтора может привести к сбою запросов с коннекторами, которые явно не поддерживают отказоустойчивое выполнение. Например, коннекторы к Kafka, Redis и многим другим NoSQL-хранилищам не поддерживают механизм отказоустойчивого выполнения. Если же коннектор поддерживает отказоустойчивое выполнение, помимо свойства retry-policy, надо еще настроить следующие параметры:
- deduplication-buffer-size – размер буфера в памяти координатора, используемого отказоустойчивым выполнением для хранения выходных данных этапов запроса. Если этот буфер заполняется во время выполнения запроса, запрос завершается с сообщением об ошибке «Exchange manager must be configure to the failure recovery capabilities to be full-function». Для эффективного восстановления после сбоя необходимо настроить диспетчер обмена (Exchange manager), который использует внешнее хранилище для буферизованных данных, позволяя хранить выгруженные данные за пределами буфера в памяти. Например, в качестве внешнего хранилища можно использовать AWS S3 и другие системы, поддерживающие S3-протокол: Azure Blob Storage, Google Cloud Storage, HDFS. О том, почему в Trino удалена поддержка доступа к Azure Storage, Google Cloud Storage, S3 и S3-совместимым файловым системам через Hive и что использовать вместо нее, читайте в нашей новой статье. Чтобы настроить диспетчер обмена, надо создать новый файл конфигурации exchange-manager.properties, поместив его в каталог /etc на координаторе и всех рабочих узлах. В этом файле надо настроить свойство exchange-manager.name, указав файловое или объектное хранилище. Чтобы снизить общую нагрузку ввода-вывода диспетчера обмена, свойство конфигурации exchange.compression-codec по умолчанию имеет значение LZ4, а сжатие и распаковка файлов выполняются автоматически. Также рекомендуется настроить правило жизненного цикла контейнера для автоматического истечения срока действия заброшенных объектов в случае сбоя узла.
- fault-tolerant-execution.exchange-encryption-enabled — шифрование данных спулинга, режима клиентского протокола с повышенной пропускной способностью, о котором мы писали здесь. Рекомендуется всегда включать это свойство, если Trino обрабатывает конфиденциальные данные. Trino шифрует данные перед тем, как поместить их в хранилище. Это предотвращает доступ к данным запроса кого-либо, кроме кластера Trino, который их записал, включая администраторов системы хранения. Новый ключ шифрования генерируется случайным образом для каждого запроса и истекает после его завершения.
- retry-initial-delay — минимальное время тайм-аута перед повтором неудавшегося запроса или задачи;
- retry-max-delay – максимальное время тайм-аута перед повтором неудавшегося запроса или задачи;
- retry-delay-scale-factor – коэффициент, на который увеличивается задержка между последующими попытками после каждой неудачи. Этот параметр полезен при нестабильных сетевых соединений или взаимодействии с внешними сервисами, где временные сбои могут быть исправлены без вмешательства пользователя. Настройка этого параметра может улучшить надежность и устойчивость работы Trino в разных инфраструктурных средах и/или с различными хранилищами данных.
В заключение отметим, что режим отказоустойчивого выполнения предлагает несколько адаптивных оптимизаций плана, которые динамически корректируют планы выполнения запросов на основе статистики времени выполнения. Это мы рассмотрим в следующий раз.
Освойте работу с Trino на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
