1. Lý do cần hợp nhất các tệp nhỏ

HDFS được thiết kế để lưu trữ hiệu quả các tệp lớn. Mỗi tệp trong HDFS đều có thông tin siêu dữ liệu riêng. Khi có quá nhiều tệp nhỏ, lượng siêu dữ liệu này sẽ tăng lên đáng kể, gây áp lực lớn lên bộ nhớ của trình quản lý HDFS.

Do đó, việc hợp nhất các tệp nhỏ thành các tệp lớn hơn là một phương pháp tối ưu hóa phổ biến và hữu ích trong các dự án.

2. Hợp nhất tệp nhỏ trên máy cục bộ và tải lên HDFS

Bạn có thể hợp nhất nhiều tệp nhỏ cục bộ thành một tệp duy nhất và sau đó tải lên HDFS bằng lệnh appendToFile của trình khách HDFS.

Giả sử chúng ta có hai tệp nhỏ trên máy cục bộ:

# Nội dung tệp user1.txt:
1,tom,male,16
2,jerry,male,10

# Nội dung tệp user2.txt:
101,jack,male,19
102,rose,female,18

Thực hiện hợp nhất:

hdfs dfs -appendToFile user1.txt user2.txt /test/upload/merged_user.txt

Nội dung tệp sau khi hợp nhất sẽ là sự kết hợp của cả hai tệp.

3. Hợp nhất tệp nhỏ trong HDFS và tải về máy cục bộ

Bạn có thể sử dụng lệnh getmerge của trình khách HDFS để hợp nhất nhiều tệp nhỏ trong HDFS thành một tệp lớn và tải về máy cục bộ.

# Tải các tệp nhỏ lên HDFS trước:
hdfs dfs -put user1.txt user2.txt /test/upload
# Tải về và hợp nhất:
hdfs dfs -getmerge /test/upload/user*.txt ./merged_user.txt

Tệp merged_user.txt tải về sẽ chứa nội dung đã được hợp nhất.

4. Thực hiện hợp nhất và tải tệp bằng Java API

Dưới đây là đoạn mã Java mẫu để thực hiện việc này:

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class HdfsFileMerger {

    public void mergeAndUploadFiles(String hdfsUri, String hdfsUser, String hdfsOutputPath, String localInputDirectory) throws URISyntaxException, IOException {
        FileSystem hdfs = FileSystem.get(new URI(hdfsUri), new Configuration(), hdfsUser);
        FSDataOutputStream hdfsOutputStream = hdfs.create(new Path(hdfsOutputPath));

        LocalFileSystem localFs = FileSystem.getLocal(new Configuration());
        Path localPath = new Path(localInputDirectory);
        FileStatus[] localFileStatuses = localFs.listStatus(localPath);

        for (FileStatus fileStatus : localFileStatuses) {
            if (fileStatus.getPath().getName().startsWith("user")) { // Lọc các tệp cần hợp nhất
                try (FSDataInputStream localInputStream = localFs.open(fileStatus.getPath())) {
                    IOUtils.copyBytes(localInputStream, hdfsOutputStream, new Configuration(), false); // Sao chép dữ liệu
                }
            }
        }

        IOUtils.closeStream(hdfsOutputStream);
        localFs.close();
        hdfs.close();
    }

    public static void main(String[] args) {
        HdfsFileMerger merger = new HdfsFileMerger();
        try {
            // Thay thế bằng thông tin HDFS và thư mục cục bộ của bạn
            merger.mergeAndUploadFiles(
                "hdfs://localhost:9000",  // URI HDFS
                "your_user",             // Tên người dùng HDFS
                "/test/upload/merged_by_java.txt", // Đường dẫn tệp HDFS đầu ra
                "/path/to/local/files"   // Đường dẫn thư mục chứa tệp nhỏ cục bộ
            );
            System.out.println("Files merged and uploaded successfully!");
        } catch (URISyntaxException | IOException e) {
            e.printStackTrace();
        }
    }
}

Kết quả hợp nhất bằng Java API sẽ tương tự như khi sử dụng lệnh.