Định nghĩa và Cấu trúc của DataNode

Lớp DataNode trong Hadoop HDFS là một thành phần quan trọng, chịu trách nhiệm lưu trữ dữ liệu thực tế. Nó được định nghĩa như sau:

public class DataNode extends Configured implements InterDatanodeProtocol,
                                                   ClientDatanodeProtocol,
                                                   FSConstants,
                                                   Runnable,
                                                   DataNodeMXBean

Từ khai báo trên, ta thấy DataNode triển khai nhiều giao diện khác nhau, phản ánh vai trò đa nhiệm của nó:

• ClientDatanodeProtocol: Giao diện này cho phép DataNode tương tác với các ứng dụng khách (Client), xử lý các yêu cầu đọc/ghi dữ liệu.
• InterDatanodeProtocol: Cho phép DataNode giao tiếp với các DataNode khác, thường là để sao chép khối dữ liệu hoặc phục hồi dữ liệu.
• Runnable: Chỉ ra rằng DataNode có thể chạy như một luồng độc lập, thực hiện các tác vụ định kỳ và xử lý yêu cầu.
• Các giao diện khác như FSConstants và DataNodeMXBean cung cấp các hằng số hệ thống tệp và khả năng giám sát qua JMX.

Một thành phần cốt lõi của DataNode là ipcServer, khởi tạo dịch vụ IPC (Inter-Process Communication). Thông qua ipcServer, DataNode có thể cung cấp các khả năng được định nghĩa trong ClientDatanodeProtocol và InterDatanodeProtocol.

Quy trình Khởi động DataNode

Quy trình khởi động của DataNode bắt đầu từ phương thức main, sau đó chuyển giao trách nhiệm chính cho phương thức startDataNode. Đây là nơi các dịch vụ và thành phần cần thiết của DataNode được khởi tạo và cấu hình.

Luồng hoạt động chính của DataNode

Khi DataNode đã khởi động, nó chạy trong một vòng lặp liên tục thông qua phương thức run(). Tuy nhiên, logic hoạt động cốt lõi nằm trong phương thức offerService(), quản lý các tương tác định kỳ với NameNode và các tác vụ khác.

Chi tiết về phương thức offerService()

Phương thức offerService() chịu trách nhiệm duy trì liên lạc với NameNode và thực hiện các hoạt động quản lý khối dữ liệu. Nó thực hiện một loạt các bước trong mỗi chu kỳ:

1. Gửi Heartbeat (Nhịp tim): DataNode kiểm tra xem đã đến lúc gửi báo cáo nhịp tim tới NameNode chưa. Nếu đã quá thời gian quy định, nó sẽ gửi một gói nhịp tim chứa thông tin về dung lượng ổ đĩa, không gian đã sử dụng, không gian còn trống, số lượng truyền dữ liệu đang diễn ra và số lượng máy thu dữ liệu (DataXceiverServer). NameNode sẽ trả về một tập hợp các lệnh để DataNode xử lý.
2. Báo cáo các khối đã nhận: DataNode thông báo cho NameNode về bất kỳ khối dữ liệu mới nào đã được nhận và lưu trữ thành công, cũng như các gợi ý về khối có thể xóa.
3. Gửi Block Report (Báo cáo khối): Định kỳ, DataNode gửi một báo cáo chi tiết về tất cả các khối dữ liệu mà nó đang lưu trữ cho NameNode. Báo cáo này giúp NameNode duy trì bản đồ vị trí khối chính xác trong toàn bộ hệ thống. NameNode cũng có thể trả về các lệnh để DataNode xử lý sau báo cáo này.
4. Khởi động Block Scanner: Nếu Block Scanner chưa được khởi động và quá trình nâng cấp (upgrade) đã hoàn tất, một luồng Block Scanner sẽ được khởi chạy để định kỳ quét và kiểm tra tính toàn vẹn của các khối dữ liệu.
5. Chế độ ngủ: Nếu không có công việc nào khác cần xử lý ngay lập tức, DataNode sẽ tạm dừng trong một khoảng thời gian nhất định, thường là cho đến chu kỳ nhịp tim tiếp theo hoặc khi có sự kiện mới (như nhận được khối dữ liệu).

Xử lý các lệnh từ NameNode (processCommand)

Khi NameNode gửi các lệnh về cho DataNode (thường thông qua phản hồi nhịp tim hoặc báo cáo khối), phương thức processCommand của DataNode sẽ giải thích và thực thi chúng. Các loại lệnh chính được định nghĩa trong DatanodeProtocol bao gồm:

• DNA_UNKNOWN = 0: Lệnh không xác định.
• DNA_TRANSFER = 1: Yêu cầu DataNode truyền một hoặc nhiều khối dữ liệu tới một DataNode khác. DataNode sẽ tạo các tác vụ truyền dữ liệu (DataTransfer) để gửi từng khối, sử dụng BlockSender để truyền dữ liệu và siêu dữ liệu mà không thực hiện kiểm tra checksum trong quá trình này.
• DNA_INVALIDATE = 2: Chỉ thị DataNode xóa các khối dữ liệu đã được đánh dấu là không hợp lệ. DataNode sẽ loại bỏ tất cả các khối được chỉ định.
• DNA_SHUTDOWN = 3: Yêu cầu DataNode ngừng hoạt động. Điều này bao gồm dừng các dịch vụ nội bộ như infoServer, DataXceiverServer, DataBlockScanner và các luồng xử lý khác, đồng thời giải phóng các thư mục lưu trữ.
• DNA_REGISTER = 4: Yêu cầu DataNode thực hiện lại quy trình đăng ký với NameNode. Điều này thường xảy ra khi DataNode bị mất kết nối với NameNode và cần thiết lập lại liên lạc.
• DNA_FINALIZE = 5: Hoàn tất quá trình nâng cấp hệ thống (upgrade). DataNode sẽ gọi phương thức finalizeUpgrade của DataStorage để hoàn thành quá trình này.
• DNA_RECOVERBLOCK = 6: Yêu cầu phục hồi một khối dữ liệu. DataNode sẽ tạo một luồng riêng để xử lý việc phục hồi từng khối, gọi phương thức recoverBlock để tái tạo thông tin khối.

Mã nguồn minh họa vòng lặp offerService()

Dưới đây là một ví dụ đơn giản hóa về cấu trúc của vòng lặp chính trong offerService():

public void offerService() throws Exception {
    while (shouldContinueRunning) {
        try {
            long currentTimestamp = System.currentTimeMillis();

            // Xử lý Heartbeat định kỳ
            if (currentTimestamp - lastHeartbeatTime > heartbeatIntervalMillis) {
                DatanodeCommand[] receivedCommands = namenodeService.sendHeartbeat(
                    datanodeRegistration,
                    storageInfo.getTotalCapacity(),
                    storageInfo.getUsedSpace(),
                    storageInfo.getRemainingSpace(),
                    activeTransfers.get(),
                    dataReceiverCount.get()
                );
                if (!processNameNodeCommands(receivedCommands)) {
                    continue; // Tiếp tục vòng lặp nếu có lệnh đặc biệt yêu cầu.
                }
                lastHeartbeatTime = currentTimestamp;
            }

            // Báo cáo các khối đã nhận thành công
            Block[] newlyReceivedBlocks;
            String[] deletionSuggestions;

            synchronized (pendingReceivedBlockList) {
                int blockCount = pendingReceivedBlockList.size();
                newlyReceivedBlocks = pendingReceivedBlockList.toArray(new Block[blockCount]);
                deletionSuggestions = pendingDeletionHints.toArray(new String[blockCount]);
                pendingReceivedBlockList.clear(); // Xóa danh sách sau khi đã báo cáo
                pendingDeletionHints.clear();
            }

            if (newlyReceivedBlocks.length > 0) {
                namenodeService.blockReceived(datanodeRegistration, newlyReceivedBlocks, deletionSuggestions);
            }

            // Gửi Block Report định kỳ
            if (currentTimestamp - lastBlockReportTime > blockReportIntervalMillis) {
                Block[] fullBlockReport = storageInfo.generateBlockReport();
                DatanodeCommand blockReportCmd = namenodeService.blockReport(
                    datanodeRegistration,
                    BlockListAsLongs.convertToArrayLongs(fullBlockReport)
                );
                processNameNodeCommands(new DatanodeCommand[]{blockReportCmd});
                lastBlockReportTime = currentTimestamp;
            }

            // Khởi động Block Scanner nếu cần
            if (blockScanner != null && scannerThread == null && upgradeManager.isUpgradeFinalized()) {
                scannerThread = new Daemon(blockScanner);
                scannerThread.start();
            }

            // Chờ cho đến chu kỳ tiếp theo hoặc khi có sự kiện
            long sleepDuration = heartbeatIntervalMillis - (System.currentTimeMillis() - lastHeartbeatTime);
            if (sleepDuration > 0 && pendingReceivedBlockList.isEmpty()) {
                try {
                    synchronized (pendingReceivedBlockList) {
                        pendingReceivedBlockList.wait(sleepDuration);
                    }
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt(); // Đặt lại trạng thái ngắt
                    // Xử lý gián đoạn nếu cần
                }
            }

        } catch (RemoteException rpcEx) {
            // Xử lý lỗi giao tiếp từ xa với NameNode
        } catch (IOException ioEx) {
            // Xử lý các lỗi I/O khác
        }
    }
}