Trino 2 — Kiến trúc Coordinator & Worker

Ở bài Trino 1 — Tổng quan chúng ta đã định vị Trino như một query engine (bộ máy truy vấn) phân tán, tách rời khỏi nơi lưu trữ dữ liệu: nó không sở hữu bảng nào, mà đọc dữ liệu tại chỗ từ nhiều nguồn (data lake, PostgreSQL, Kafka…) và ghép lại bằng SQL chuẩn ANSI. Bài này đi sâu vào cách Trino biến một câu SELECT thành hàng trăm đơn vị công việc chạy song song trên nhiều máy — kiến thức nền tảng để sau này bạn tune hiệu năng, đọc EXPLAIN, và chọn kích cỡ cụm cho đúng.

Trino thuộc họ MPP (Massively Parallel Processing — xử lý song song quy mô lớn): thay vì một tiến trình khổng lồ, hàng chục process trên nhiều node cùng gặm một phần dữ liệu, rồi kết quả được ghép lại. Điểm mấu chốt khiến Trino nhanh (nhưng cũng "ăn" RAM) là dữ liệu chảy giữa các tầng xử lý trong bộ nhớ, dạng streaming, không ghi ra đĩa như MapReduce đời cũ.

Hai loại node: Coordinator vs Worker

Một cụm Trino gồm đúng hai vai trò tiến trình, chạy cùng một binary nhưng khác cấu hình (coordinator=true/false trong config.properties):

Coordinator giống "nhạc trưởng": không chơi nhạc cụ nào nhưng quyết định ai chơi phần gì, khi nào. Worker là "nhạc công". Trong cụm nhỏ có thể để một node vừa làm coordinator vừa làm worker, nhưng production luôn tách riêng để coordinator không bị nghẽn.

Client và Trino protocol

Client (Trino CLI, driver JDBC/ODBC, các công cụ BI như Superset/Metabase, hay code Python qua thư viện) nói chuyện với coordinator qua giao thức Trino — về bản chất là một loạt lời gọi HTTP/REST. Khi client gửi SQL, coordinator trả về một queryId cùng một URI nextUri; client poll (hỏi vòng) liên tục vào URI đó để lấy từng lô kết quả (dữ liệu chảy về dần theo dạng phân trang), cho đến khi trạng thái là FINISHED. Nhờ cơ chế streaming này, kết quả có thể bắt đầu về ngay khi các stage đầu tiên sinh ra dữ liệu, không phải chờ toàn bộ query xong.

Discovery service

Worker cần biết coordinator ở đâu và coordinator cần biết có những worker nào đang sống. Việc đó do discovery service đảm nhiệm — mặc định nhúng ngay trong coordinator (không cần dịch vụ ngoài). Mỗi worker khi khởi động sẽ đăng ký (discovery.uri trỏ về coordinator) và gửi heartbeat định kỳ; worker chết sẽ bị loại khỏi danh sách lập lịch. Đây là lý do khi scale-out bạn chỉ cần bật thêm worker trỏ đúng discovery.uri, cụm tự nhận diện.

Mô hình thực thi: từ SQL đến split

Đây là phần cốt lõi. Coordinator biến một câu SQL qua nhiều tầng biểu diễn:

1. Query → Logical plan (kế hoạch logic): cây các phép toán quan hệ (scan, filter, join, aggregate, sort…) chưa quan tâm đến máy nào chạy. Tại đây bộ tối ưu (cost-based optimizer nếu có thống kê) sắp lại thứ tự join, đẩy filter xuống gần nguồn (predicate pushdown), cắt cột thừa (projection pushdown).
2. Distributed plan (kế hoạch phân tán): logical plan được cắt thành các STAGE. Ranh giới stage nằm ở chỗ dữ liệu buộc phải tái phân phối giữa các node — ví dụ trước một join hash hay một aggregate theo khóa, dữ liệu phải được "băm" (shuffle) sao cho mọi dòng cùng khóa về cùng một worker.
3. Mỗi STAGE được nhân bản thành nhiều TASK — mỗi task chạy trên một worker. Một stage có 10 worker thì có ~10 task cùng làm phần việc của stage đó.
4. Mỗi TASK xử lý nhiều SPLIT. Split là đơn vị dữ liệu song song nhỏ nhất — ví dụ một khối (block) trong file Parquet, một range của bảng, một partition. Task lấy split, chạy chúng qua một pipeline of OPERATORS (chuỗi toán tử: TableScan → Filter → Project → …), khai thác song song ở mức nhiều luồng (driver) bên trong.

Tóm gọn thứ bậc: Query ⊃ Stage ⊃ Task ⊃ Split ⊃ Operator pipeline. Khi đọc output EXPLAIN (TYPE DISTRIBUTED) bạn sẽ thấy đúng cây stage này, kèm nhãn kiểu phân phối (SINGLE, HASH, BROADCAST).

Exchange & shuffle: in-memory streaming

Dữ liệu giữa hai stage đi qua exchange. Có nhiều kiểu:

• Broadcast: bảng nhỏ được nhân bản tới mọi worker (dùng cho broadcast join — bảng nhỏ nằm gọn trong RAM mỗi node).
• Hash / partitioned (shuffle): dữ liệu được băm theo khóa join/group để dồn về đúng worker.
• Gather: gom về một node (stage cuối trước khi trả client).

Điểm quyết định hiệu năng: theo mặc định (chế độ truyền thống), exchange là streaming trong bộ nhớ — output của một stage được đẩy thẳng qua mạng vào stage kế, không materialize (không ghi) ra đĩa. Đây là lý do Trino nhanh hơn nhiều so với Hive/MapReduce (vốn ghi tạm ra HDFS giữa mỗi bước). Cái giá là: cả pipeline phải cùng chạy đồng thời và giữ dữ liệu trung gian trong RAM; nếu một worker chết giữa chừng, không có checkpoint để nối lại → cả query hỏng và phải chạy lại từ đầu.

Vai trò của Connector (SPI)

Trino kết nối tới mọi nguồn qua connector, hiện thực giao diện SPI (Service Provider Interface). Một connector cung cấp cho engine ba thứ:

1. Metadata: danh sách schema/bảng/cột, kiểu dữ liệu, thống kê (số dòng, số giá trị phân biệt) để tối ưu hóa.
2. Data location + splits: connector chịu trách nhiệm liệt kê split — nó biết dữ liệu nằm ở đâu (file nào trên S3, partition nào) và cắt thành các split để worker đọc song song. Đây là mấu chốt của mức song song hóa: một bảng Parquet 200 file có thể sinh hàng trăm split.
3. Data access + pushdown: đọc dữ liệu và, nếu nguồn hỗ trợ, đẩy bớt filter/aggregate xuống nguồn (predicate/aggregate pushdown) để giảm dữ liệu chuyển qua mạng.

Vì cùng một giao diện SPI, engine không cần biết bên dưới là Hive, Iceberg, PostgreSQL hay MongoDB — chi tiết ở bài Trino 3 — Connectors và cách ghép nhiều nguồn ở Trino 4 — SQL Federation.

Song song hóa & phân phối

Mức song song của một query = (số worker) × (số split đọc song song mỗi worker) × (số luồng operator). Ba đòn bẩy để tăng thông lượng:

• Nhiều worker hơn → nhiều task song song hơn cho mỗi stage.
• Nhiều split hơn (định dạng file chia được, partition hợp lý) → tận dụng hết worker; ngược lại một file khổng lồ không chia được sẽ để nhiều worker "ngồi chơi".
• task.concurrency điều chỉnh số driver song song trong một task.

Phân phối dữ liệu (data skew — lệch dữ liệu) là kẻ thù: nếu một khóa join chiếm 80% số dòng, worker giữ khóa đó thành nút thắt cổ chai. Xử lý skew được bàn kỹ ở Trino 6 — Performance Tuning.

Quản lý bộ nhớ

Vì exchange chạy trong RAM, quản lý bộ nhớ là chuyện sống còn. Trino chia bộ nhớ thành:

• User memory: bộ nhớ trực tiếp phục vụ dữ liệu người dùng — hash table của join, buffer của aggregate/sort. Giới hạn bởi query.max-memory-per-node (mỗi node) và query.max-memory (toàn cụm cho một query).
• System memory: bộ nhớ engine dùng cho buffer đọc, exchange…

Khi một query vượt giới hạn, nó bị kill (giết) để bảo vệ cụm — bạn sẽ thấy lỗi EXCEEDED_LOCAL_MEMORY_LIMIT. Có hai lối thoát: (a) bật spill-to-disk — cho phép join/aggregate/sort tràn dữ liệu ra đĩa cục bộ khi hết RAM (chậm hơn nhưng không chết query); (b) dùng resource group để phân bổ hạn mức theo nhóm người dùng, tránh một query "nuốt" cả cụm.

Fault tolerance: truyền thống vs Tardigrade

Đây là khác biệt kiến trúc lớn nhất giữa hai chế độ chạy của Trino:

Nói cách khác, Tardigrade đánh đổi một chút độ trễ để lấy độ tin cậy: một job ETL 45 phút không còn phải chạy lại toàn bộ chỉ vì một worker bị OOM ở phút thứ 40. Với ngân hàng, đây là lựa chọn mặc định cho các pipeline nạp dữ liệu ban đêm.

Cluster sizing khái quát

Không có công thức tuyệt đối, nhưng các nguyên tắc thực chiến:

• Coordinator: 1 node đủ mạnh (CPU nhiều core để plan/optimize, RAM vừa phải). Coordinator quá tải thường do quá nhiều query đồng thời hoặc kết quả trả về khổng lồ — cân nhắc coordinator HA.
• Worker: đồng nhất cấu hình, RAM là tài nguyên quý nhất (vì exchange in-memory). Quy tắc thô: tổng RAM worker ≥ dữ liệu trung gian lớn nhất của query nặng nhất, cộng dư địa. Cấp -Xmx (heap JVM) khoảng 70–80% RAM máy.
• Số worker: scale theo mức song song mong muốn và SLA. Bắt đầu từ ước lượng dữ liệu quét / thông lượng mỗi worker, rồi đo và điều chỉnh.
• Autoscaling: dùng discovery service để thêm/bớt worker động (ví dụ trên Kubernetes — xem Trino 8 — Deployment & Ops).

Đọc kế hoạch bằng EXPLAIN

EXPLAIN là cửa sổ nhìn vào toàn bộ cơ chế trên. Trên PostgreSQL sandbox bạn có thể chạy EXPLAIN để làm quen khái niệm plan (dù cú pháp và stage khác Trino):

-- ▶ Chạy được
EXPLAIN SELECT currency, SUM(balance) FROM accounts GROUP BY currency;

Trên Trino (cú pháp riêng, KHÔNG chạy trong sandbox — chỉ minh họa), bạn xem kế hoạch phân tán để biết engine chia bao nhiêu stage và exchange kiểu gì:

-- Trino — minh họa, KHÔNG chạy trong sandbox PostgreSQL
EXPLAIN (TYPE DISTRIBUTED)
SELECT c.city, count(*) AS n
FROM lakehouse.core.transactions t
JOIN lakehouse.core.accounts a  ON t.account_id = a.id
JOIN lakehouse.core.customers c ON a.customer_id = c.id
WHERE t.amount > 1000000000
GROUP BY c.city;

Output (rút gọn, minh họa) cho thấy cây stage: Fragment 0 [SINGLE] gom kết quả về coordinator; Fragment 1 [HASH] chạy aggregate theo city; các fragment dưới ScanFilter đọc transactions/accounts/customers, nối bằng RemoteExchange kiểu HASH hoặc REPLICATE (broadcast). Đọc được cây này chính là biết query của bạn sẽ shuffle ở đâu và tốn RAM chỗ nào.

Use case thực tế

Bối cảnh (NCB): Khối Phân tích Dữ liệu vận hành một cụm Trino phục vụ truy vấn phân tích liên hợp. Cụm gồm 1 coordinator + 12 worker, mỗi worker 64 GB RAM (heap JVM -Xmx=50g). Nguồn chính là lakehouse Iceberg trên object storage (~2 tỷ dòng giao dịch/tháng), ghép thêm bảng danh mục sản phẩm nằm trên PostgreSQL.

Sự cố ban đầu: Một báo cáo AML (chống rửa tiền) hằng đêm — quét 90 ngày giao dịch, join với danh mục khách hàng rủi ро cao rồi aggregate theo tỉnh/thành — chạy mất ~40 phút và fail khoảng 3 lần/tuần vì đúng một worker bị OOM ở giai đoạn shuffle, kéo cả query chết và phải chạy lại từ đầu, đẩy job trễ giờ bàn giao.

Xử lý, theo đúng kiến trúc bài này:

1. Chạy EXPLAIN (TYPE DISTRIBUTED) → phát hiện một broadcast join sai (bảng "nhỏ" thực ra 8 GB đang bị nhân bản tới mọi worker), và một stage aggregate bị skew theo tỉnh (Hà Nội chiếm ~35% dòng).
2. Ép chuyển broadcast → partitioned (hash) join để bảng lớn không nhân bản; bật thống kê để optimizer chọn thứ tự join tốt hơn.
3. Bật spill-to-disk cho aggregate để chống OOM cục bộ.
4. Với chính job ETL này, bật Fault-Tolerant Execution (retry-policy=TASK) để mỗi task lỗi được retry thay vì giết cả query.

Kết quả: thời gian chạy còn ~18 phút, và trong 6 tuần sau đó không job nào phải chạy lại từ đầu — task lỗi lẻ được engine tự retry. Đổi lại độ trễ tăng nhẹ vài phút do exchange materialize, nhưng hoàn toàn chấp nhận được cho batch ban đêm.

Ghi nhớ

• Trino là engine MPP: coordinator lập & tối ưu kế hoạch rồi điều phối (không xử lý dữ liệu bảng); worker thực thi song song và đọc dữ liệu qua connector.
• Thứ bậc thực thi: Query → Stage → Task → Split → Operator pipeline. Split là đơn vị dữ liệu song song nhỏ nhất; connector là bên tạo ra split.
• Discovery service (nhúng trong coordinator) giúp worker đăng ký và cụm tự nhận diện khi scale-out.
• Exchange/shuffle giữa stage mặc định là in-memory streaming, không ghi đĩa → nhanh nhưng tốn RAM và query hỏng phải chạy lại từ đầu.
• Connector (SPI) cung cấp metadata + data location + tạo split + pushdown; nhờ nó engine tách rời khỏi storage.
• Bộ nhớ chia user/system; vượt hạn mức thì query bị kill — cứu bằng spill-to-disk và resource group.
• Fault-Tolerant Execution (Tardigrade): materialize exchange ra spooling storage, retry theo task; đánh đổi độ trễ lấy độ bền — lý tưởng cho ETL/batch dài.
• Dùng EXPLAIN (TYPE DISTRIBUTED) để đọc cây stage, phát hiện broadcast sai và data skew — bước đầu tiên khi tune (xem Trino 6). Mô hình stage/shuffle này rất giống Spark SQL — đối chiếu ở spark-03-spark-sql.