MIT 6.5840: Lab 01 - MapReduce

OverviewCopy Anchor Link

Lab 01 yêu cầu viết một hệ thống MapReduce phân tán, gồm một process coordinator làm nhiệm vụ quản lý và giao task, cùng nhiều process worker chạy song song để thực hiện các task Map và Reduce. Các process này giao tiếp với nhau qua RPC trên unix socket, và kết quả cuối cùng phải giống hệt kết quả của một bản chạy tuần tự mrsequential.go được cung cấp.

Bài lab cung cấp sẵn một số MapReduce app trong src/mrapps/ dùng làm test case. Mỗi app chỉ gồm hai hàm Map và Reduce, được build thành shared object (file .so), và chỉ được worker nạp vào bằng gói plugin của Go khi chương trình thực sự chạy. Cách làm này khá giống Hadoop MapReduce, nơi người dùng đóng gói app thành file .jar rồi submit file đó cho framework.

Bên cạnh đó, lab có sẵn mrsequential.go, một bản MapReduce chạy tuần tự trên một process duy nhất: đọc file, gọi Map, sort, gọi Reduce rồi ghi output. Đây vừa là bản tham khảo để hiểu luồng xử lý, vừa là thứ mà test dùng để sinh ra đáp án đúng.

Map/Reduce ContractCopy Anchor Link

Mỗi framework định nghĩa signature của Map và Reduce theo một cách riêng.

Paper gốc của Google chỉ mô tả ở mức khái niệm, không quy định kiểu dữ liệu hay cách implement cụ thể:

map    (k1, v1)        → list(k2, v2)
reduce (k2, list(v2))  → list(v2)

Hadoop MapReduce cụ thể hóa ý tưởng này bằng hai class generic Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> và Reducer<...>. Người dùng tự chọn kiểu cụ thể khi kế thừa, rồi override các method map và reduce tương ứng.

Riêng bài Lab này thì đơn giản hóa tối đa khi mọi thứ đều là string. Mỗi app là một plugin Go, build bằng go build -buildmode=plugin và bắt buộc phải export đúng hai hàm với đúng tên và đúng kiểu:

func Map(filename string, contents string) []mr.KeyValue
func Reduce(key string, values []string) string

Tên tham số có thể đặt tùy ý, nhưng tên hàm và kiểu thì phải khớp chính xác. Lý do là worker nạp plugin lúc chạy bằng plugin.Open, tìm hàm theo tên bằng Lookup("Map") và Lookup("Reduce"), rồi ép kiểu về đúng signature trên.

Khi sử dụng framework, users chỉ cần viết logic bên trong hai hàm này. Họ vẫn có thể thêm hàm phụ, dùng biến toàn cục hay import thư viện tùy ý, còn mọi việc khác do framework lo.

1. Framework đọc từng file input và gọi Map(filename, contents) một lần cho mỗi file.
2. Sau đó nó gom tất cả cặp {Key, Value} mà Map sinh ra, sort và nhóm chúng theo key, rồi gọi Reduce(key, values[]) một lần cho mỗi key khác nhau.
3. Cuối cùng, kết quả được ghi ra các file mr-out-*, mỗi dòng theo định dạng "%v %v\n", tức key output.

Framework tự lo việc chạy lại task khi worker chết: nó ghi output ra file tạm rồi mới đổi tên, nên file kết quả không bao giờ bị ghi dở, và dù một task chạy mấy lần thì cũng chỉ giữ lại một bản. Nhưng để các lần chạy cho ra cùng một kết quả, người viết app cần để ý hai thứ.

• Thứ nhất, thứ tự các phần tử trong values truyền vào Reduce không cố định, vì nó phụ thuộc vào cách framework gom dữ liệu chứ không phải do Map function. Vì vậy Reduce phải cho cùng kết quả với mọi thứ tự: hoặc tự sort trước như indexer, hoặc chỉ dùng phép tính không quan tâm thứ tự như đếm số phần tử trong wc.
• Thứ hai, Map và Reduce có thể bị gọi lại nhiều lần với cùng input, nên chúng không được dùng số ngẫu nhiên hay thời gian hiện tại để quyết định output.

What the Lab ProvidesCopy Anchor Link

Lab đã dựng sẵn toàn bộ phần khung bên ngoài: entrypoint của hai loại process, hạ tầng RPC, vài utils và bộ test. Khung này cũng đã ngầm định sẵn vai trò như sau:

• Worker trực tiếp nạp plugin hay MR app. Code của user không được gửi đi đâu cả. File .so nằm sẵn trên ổ đĩa dùng chung, và worker tự mở nó ra theo arg được truyền qua câu lệnh khởi chạy.
• Coordinator sẽ quản lý danh sách task, đồng thời đóng vai RPC server, nhận các RPC xin và báo hoàn thành task từ worker. Phần quản lý này mình phải tự implement.
• Worker chủ động gọi RPC tới coordinator để xin task và báo khi xong việc.
• Dữ liệu được shuffle qua local disk: map worker ghi các cặp key/value ra file trung gian mr-X-Y theo bucket ihash(key) % nReduce, còn reduce worker Y đọc tất cả file mr-*-Y để gom các key thuộc về mình.
• Định kì mỗi giây, Coordinator sẽ gọi lại function Done() để biết khi nào job kết thúc. Việc định nghĩa function này là của bạn.

What We Need to BuildCopy Anchor Link

Bài lab đã dựng sẵn entrypoint và hạ tầng RPC, nhưng để trống phần điều phối task và thực thi MapReduce. Mình cần tự định nghĩa coordinator giữ những state gì, worker trao đổi với coordinator qua RPC ra sao, và mỗi task đọc/ghi dữ liệu thế nào.

Checklist của tớ là:

• Trong MakeCoordinator, tạo một map task cho mỗi file input và nReduce reduce task; lưu trạng thái cùng thời điểm giao của từng task. Vì các RPC handler có thể chạy đồng thời, state chung cần được bảo vệ bằng sync.Mutex.
• Định nghĩa RPC để worker xin task và báo hoàn thành. Khi worker xin việc, coordinator giao một task chưa chạy, bảo chờ nếu hiện chưa có việc, hoặc bảo thoát khi cả job đã xong. Reduce task chỉ được giao sau khi tất cả map task hoàn thành.
• Nếu một task đã giao quá khoảng 10 giây mà chưa hoàn thành, coordinator cho phép giao lại task đó. Vì worker cũ có thể vẫn đang chạy và báo xong muộn, handler báo hoàn thành phải xử lý được các lần báo trùng hoặc đến muộn.
• Worker lặp lại việc xin task, chạy Map hoặc Reduce rồi báo kết quả. Map task ghi các file trung gian mr-X-Y; reduce task đọc các file thuộc partition của mình và tạo mr-out-Y. File được ghi vào đường dẫn tạm trước khi rename sang tên cuối để tránh lộ file đang ghi dở.
• Done() trả về true khi toàn bộ reduce task đã hoàn thành, để entrypoint của coordinator biết lúc nào có thể thoát.

ImplementationCopy Anchor Link

Step 0: TaskCopy Anchor Link

Questions:

• Có mấy loại task?
• Trạng thái?
• Định danh?

Appendix: From One Machine to ManyCopy Anchor Link

Phần này nằm ngoài yêu cầu của lab, nhưng đáng để suy nghĩ. Hệ thống mà lab yêu cầu chỉ chạy được trên một máy duy nhất. Coordinator và worker là các process riêng biệt, nhưng chúng ngầm dựa vào việc dùng chung một hệ điều hành và một ổ đĩa. Nếu muốn đưa nó lên một cluster thật, có ít nhất ba thứ đang bị gắn chặt vào một máy, và file .so chỉ là một trong số đó.

RPC over Unix SocketsCopy Anchor Link

Coordinator đang lắng nghe bằng net.Listen("unix", sockname), còn worker kết nối bằng rpc.DialHTTP("unix", coordSockName). Unix socket thực chất là một file trên đĩa local, nên chỉ các process trong cùng một máy mới nói chuyện được với nhau. Muốn chạy trên nhiều máy thì phải chuyển sang TCP, với địa chỉ IP và port của coordinator. Lab thậm chí đã để sẵn dòng rpc.DialHTTP("tcp", "127.0.0.1"+":1234") trong worker.go, chỉ là đang comment lại. Đây là phần dễ sửa nhất, chỉ vài dòng code.

Data: Input, Intermediate Files, and OutputCopy Anchor Link

Đây mới là vấn đề lớn nhất. Trong lab, mọi process cùng đọc và ghi một thư mục. Khi tách ra nhiều máy, worker ở máy B sẽ không đọc được file input nằm ở máy A, và reduce worker ở máy C cũng không đọc được file mr-3-5 mà map worker ở máy B vừa ghi ra.

Paper của Google giải quyết vấn đề này bằng cách đặt input và output trên GFS, một filesystem phân tán mà mọi máy đều nhìn thấy. Dữ liệu trung gian thì được xử lý khác: map worker ghi nó ra đĩa local của chính mình, sau đó reduce worker gọi RPC tới từng map worker để kéo phần dữ liệu thuộc về mình. Hadoop làm tương tự với HDFS và bước shuffle. Dữ liệu trung gian cố tình không được đưa lên filesystem chung, vì nó chỉ là dữ liệu tạm nhưng lại rất lớn, nên ghi local sẽ nhanh hơn nhiều. Cái giá phải trả là nếu một máy map chết sau khi đã hoàn thành task, file trung gian trên đĩa của nó cũng mất theo, và map task đó phải chạy lại dù trước đó đã xong. Trong lab, chuyện này không xảy ra vì mọi process dùng chung một đĩa.

Shipping User Code to WorkersCopy Anchor Link

Phần cuối là file .so. Cách đơn giản nhất là cho mọi máy mount chung một filesystem, chẳng hạn NFS, S3 mount hay GFS, rồi đặt file .so lên đó. Nếu đã phải có filesystem chung cho dữ liệu thì đặt luôn plugin lên đó là hợp lý. Cách thứ hai là tự gửi file đi: coordinator trả nội dung file .so qua RPC, worker ghi ra đĩa local rồi mới gọi plugin.Open. Cách này làm được nhưng ít ai chọn khi đã có sẵn filesystem chung. Cách thứ ba là bỏ hẳn plugin, build Map và Reduce chung vào binary của worker, rồi phân phối binary đó lên các máy bằng Docker image, Ansible hoặc hệ thống quản lý cluster.

Dù chọn cách nào, vẫn phải nhớ ràng buộc của Go plugin. File .so phải được build đúng cho OS và kiến trúc CPU của máy sẽ chạy nó, cùng phiên bản Go, cùng phiên bản package mr với worker. Một cluster có cả máy x86 lẫn ARM sẽ cần hai bản .so khác nhau. Đây là chỗ Java và Hadoop có lợi thế, vì bytecode trong file .jar chạy được trên mọi máy có JVM.

How Hadoop and Google Ship CodeCopy Anchor Link

Trong Hadoop YARN, máy client chạy lệnh hadoop jar sẽ load jar để cấu hình job và chia input thành các split, sau đó upload jar lên thư mục staging trên HDFS. ResourceManager, master của cả cluster, không hề load jar. Nó chỉ là bộ lập lịch tài nguyên chung cho mọi loại job, giống vai trò coordinator trong lab. Mỗi job còn có một MRAppMaster riêng làm nhiệm vụ theo dõi và giao lại task. Cuối cùng, NodeManager trên từng máy tự kéo jar từ HDFS về đĩa local, gọi là localization, rồi khởi động JVM với jar đó trong classpath để chạy task. Như vậy jar có đi qua mạng, nhưng không phải master gửi trực tiếp cho worker: client đẩy lên HDFS và worker tự kéo về. Việc gửi code chứ không gửi dữ liệu là theo nguyên tắc "move computation to data". Jar chỉ vài MB trong khi dữ liệu có thể lên tới hàng TB, nên Hadoop ưu tiên chạy task trên chính máy đang giữ block dữ liệu đó.

Paper gốc của Google lại làm theo cách khác. Chương trình của người dùng được link luôn với thư viện MapReduce thành một binary duy nhất. Thư viện này khởi động nhiều bản sao của chương trình trên cluster, và một bản sao trong đó đóng vai master. Nghĩa là ở Google, master và worker chạy cùng một binary, nên master cũng chứa code Map/Reduce của người dùng, chỉ là không bao giờ gọi tới. Cách này đơn giản vì không cần cơ chế plugin hay classloader, nhưng mỗi job lại là một binary riêng cần được phân phối lên cluster.

The Coordinator Is Still a Single Point of FailureCopy Anchor Link

Kể cả khi đã giải quyết hết các vấn đề trên, hệ thống vẫn còn một điểm yếu: nếu coordinator chết, cả job chết theo. Paper của Google chấp nhận điều này và chỉ đề cập tới việc định kỳ lưu checkpoint state của master. Lab cũng không yêu cầu xử lý trường hợp này. Làm cho một thành phần như master chịu được lỗi bằng cách nhân bản nó lên nhiều máy chính là bài toán mà Raft giải quyết, và đó là nội dung của các lab tiếp theo.

Shared Storage and Why the Lab Runs on One MachineCopy Anchor Link

Nhìn lại cả ba hệ thống, có thể thấy MapReduce không bao giờ đứng một mình mà luôn đi kèm một hệ thống storage chung. Ở Google, đó là GFS (Google File System), một filesystem phân tán nội bộ được công bố trong paper riêng năm 2003, một năm trước paper MapReduce. Lưu ý là GFS khác với GCS (Google Cloud Storage), một dịch vụ cloud ra đời muộn hơn nhiều. Hadoop thì là bản mã nguồn mở mô phỏng lại đúng cặp đôi này: HDFS là bản clone của GFS, còn Hadoop MapReduce là bản clone của MapReduce. Tuy vậy, storage chung chỉ lo phần đầu vào và đầu ra. Dữ liệu trung gian ở giữa vẫn nằm trên đĩa local của map worker, và MapReduce tự vận chuyển nó qua mạng tới reduce worker.

Lab này không có storage phân tán nào cả, vì mục tiêu của nó là logic của MapReduce chứ không phải hạ tầng. Mọi thứ chạy trên một máy, và ổ đĩa của máy đó đóng vai storage chung cho cả input, dữ liệu trung gian lẫn output. Dù vậy, lab vẫn cố tình tách coordinator và worker thành các process riêng thay vì các goroutine trong cùng một chương trình. Lý do không chỉ là để chạy song song thật, mà quan trọng hơn là để các thành phần có thể chết độc lập với nhau. Khi crash.go gọi os.Exit(1), chỉ đúng một worker chết, còn coordinator và các worker khác vẫn chạy bình thường. Một process chết như vậy mô phỏng khá sát chuyện một máy chết trong cluster thật. Nếu dùng goroutine, os.Exit sẽ giết cả chương trình, và lab không còn cách nào kiểm tra khả năng chịu lỗi của hệ thống.

Nói cách khác, lab giữ lại hai đặc tính cốt lõi của một hệ phân tán là chạy song song thật và lỗi độc lập giữa các thành phần. Những thứ còn lại, gồm mạng thật, storage phân tán và việc phân phối code, đều được thay bằng một máy với ổ đĩa dùng chung.

Tóm lại, lab cố tình gói mọi thứ vào một máy để mình tập trung vào phần cốt lõi là điều phối task, xử lý worker chết và ghi file nguyên tử. Muốn chạy thật trên nhiều máy, cần đổi RPC sang TCP, có filesystem chung cho input và output, có cơ chế kéo dữ liệu trung gian giữa các worker, và có cách đưa code người dùng tới các máy. Trong số đó, file .so lại là phần dễ giải quyết nhất.