Tasks и drivers: pipeline-исполнение
Из урока про таксономию: task — runtime-исполнение стадии на воркере, driver — наименьшая единица параллелизма, физическая цепочка операторов. Теперь спустимся внутрь воркера и разберём, как именно task работает: как она распадается на драйверы, что значит «pipeline-исполнение» и почему параллелизм меряется драйверами, а не задачами.
Это самый «железный» уровень после Page и Block — то, что реально крутится на ядрах процессора воркера.
Что происходит внутри задачи
Task пришла на воркер. Coordinator прислал ей описание: какой фрагмент плана исполнять и какие splits обрабатывать. Внутри воркера task должна превратить это в реальную работу на нескольких ядрах.
Task не исполняет фрагмент в одиночку. Она разворачивает его в pipeline — конвейер операторов — и запускает несколько драйверов, каждый из которых прогоняет данные через копию этого конвейера. Структура такая:
Pipeline — это шаблон: описание, из каких операторов и в каком порядке состоит цепочка. Driver — экземпляр этого шаблона: реальная цепочка операторов в памяти, которая прогоняет конкретные данные. Один pipeline — много драйверов. Драйверы одной задачи устроены одинаково (один шаблон), но работают независимо, каждый над своим куском данных.
Driver — наименьшая единица параллелизма
Повторим и углубим определение из урока про таксономию, потому что оно центральное.
Driver — это последовательность экземпляров операторов в памяти, с одним входом и одним выходом. И driver — наименьшая единица параллелизма в Trino.
Почему именно driver, а не task. Task — это «вся работа стадии на этом воркере». Но «вся работа на воркере» — не атомарна: воркер многоядерный, и одна task должна загрузить несколько ядер. Делает она это, запуская несколько драйверов. Каждый driver — отдельная независимая единица работы, которую планировщик воркера может поставить на отдельное ядро. Атомом параллелизма является driver: ниже него уже неделимо (операторы внутри driver работают строго последовательно — данные идут по цепочке).
Параллелизм, таким образом, двухуровневый. Уровень stage -> task — это параллелизм по машинам: одна стадия даёт много задач на разных воркерах. Уровень task -> driver — параллелизм по ядрам: одна задача даёт много драйверов на ядрах воркера. А driver — атом: ниже него уже неделимо.
Отсюда формула реального параллелизма запроса: сумма по всем стадиям от (число task в стадии, умноженное на число drivers в каждой task). Если у стадии 10 task и в каждой по 8 drivers — стадия исполняется в 80 параллельных потоков. Считать только задачи (10) — занизить параллелизм в 8 раз.
Число драйверов на задачу регулируется. За него отвечает настройка task.concurrency — она задаёт степень параллелизма внутри задачи (типично — близко к числу ядер воркера). Тонкая настройка параллелизма — тема отдельного модуля про планировщик; здесь важно понять принцип: драйверов на задачу несколько, и это и есть параллелизм по ядрам.
Pipeline-исполнение: данные текут, не материализуются
Теперь — как driver обрабатывает данные. Ключевое слово — pipeline (конвейер, потоковая обработка).
Driver — цепочка операторов. Данные проходят её насквозь: первый оператор берёт порцию данных, обрабатывает, отдаёт второму; второй обрабатывает, отдаёт третьему; и так до конца цепочки. Порция данных — это Page (колоночный батч, ему посвящён отдельный урок). Driver работает Page за Page: взял Page, прогнал через всю цепочку операторов, взял следующий.
Принципиально важно, чего driver не делает. Он не материализует промежуточные результаты целиком. Оператор filter не ждёт, пока scan прочитает всю таблицу, чтобы потом начать фильтровать. Как только scan выдал первый Page, filter уже его обрабатывает, а scan параллельно готовит следующий. Данные текут по конвейеру, а не накапливаются между операторами.
Преимущество потоковой обработки — в эффективности по памяти и по латентности. По памяти: в каждый момент в работе лишь несколько Page-ов, а не вся таблица целиком — driver не держит в памяти весь промежуточный результат. По латентности: первые строки результата могут пойти к клиенту, как только первые Page прошли конвейер, — не нужно ждать обработки всех данных.
Pipeline-исполнение объясняет распространённое заблуждение о скорости Trino. Trino быстр НЕ потому, что грузит данные в память и кэширует. Он быстр потому, что данные стримятся через конвейеры операторов: между операторами почти ничего не материализуется, работа идёт батчами, несколько драйверов жмут параллельно на ядрах. Скорость — от MPP-распределённости и pipelined-исполнения, а не от хранения данных в памяти.
Не все операторы одинаково «текучи». Есть блокирующие операторы — которым по природе нужны все входные данные, прежде чем выдать первую строку результата. Например, сортировка (ORDER BY) не может выдать первую строку, не увидев всех строк, — иначе непонятно, какая первая. Финальная агрегация без GROUP BY тоже: count(*) готов, только когда посчитаны все строки. Такие операторы — точки, где конвейер вынужденно «копит» данные. Но даже вокруг них остальная часть плана работает потоково.
Кооперативная многозадачность воркера
Последняя деталь «до железа». Драйверов на воркере может быть больше, чем ядер: десятки задач, у каждой несколько драйверов. Как воркер делит ядра между ними честно.
Trino использует кооперативную многозадачность. Driver работает не до полного завершения, а квантами: он обрабатывает данные ограниченное время (квант порядка долей секунды), после чего сам уступает ядро — возвращает управление планировщику воркера. Планировщик ставит на ядро следующий готовый driver. Отработавший driver встанет в очередь и продолжит, когда снова получит квант.
Это и есть «cooperative»: driver добровольно уступает ядро, не дожидаясь, пока его вытеснят. Зачем так. Если бы длинный driver занимал ядро, пока не доделает весь свой split, короткие быстрые запросы застревали бы за ним в очереди. Квантование выравнивает: ни один driver не монополизирует ядро, и сотни конкурентных запросов делят воркер плавно, без того чтобы тяжёлый запрос заморозил лёгкие.
Картина уровня task-driver целиком
Соберём. Task приходит на воркер с фрагментом плана и splits. Task разворачивает фрагмент в pipeline — шаблон цепочки операторов — и запускает несколько drivers, экземпляров этого конвейера; число драйверов задаёт параллелизм внутри воркера (task.concurrency). Каждый driver потоково прогоняет Page за Page через свою цепочку операторов, не материализуя промежуточные результаты. Драйверы делят ядра воркера через кооперативное квантование — каждый работает квант и уступает ядро.
Это и есть «рабочая лошадка» Trino. Stage — концепция, task — её воплощение на воркере, а driver — то, что в реальности молотит данные на ядре процессора.
Попробуй сам
Уровень task-driver виден в Web UI и EXPLAIN ANALYZE:
- Выполните
EXPLAIN ANALYZE SELECT orderstatus, count(*) FROM tpch.sf10.orders GROUP BY orderstatus. Откройте этот запрос в Web UI, страница деталей. Для одной стадии найдите число task и счётчики драйверов (часто в pipeline-статистике). - Прикиньте реальный параллелизм стадии: число task умножить на число drivers в задаче. Сравните с числом ядер на воркерах кластера.
- Найдите в плане оператор сортировки или финальной агрегации. Подумайте, почему он блокирующий — почему он не может работать чисто потоково.
- Сформулируйте письменно: что такое pipeline-исполнение, почему driver — наименьшая единица параллелизма, и зачем драйверы уступают ядро квантами вместо работы до конца.