实现大于 10MB 的文件流上传到 HDFS，小于的传到Hbase，使用 HDFS API的方式

提问区
1

要实现大于 10MB 的文件流上传到 HDFS，小于的传到 HBase，您可以使用 HDFS API 和 HBase API 来完成。以下是一个示例的 Java 代码，演示了如何根据文件大小选择不同的上传方式：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.Files;

public class FileUpload {
    private static final long MAX_HDFS_UPLOAD_SIZE = 10 * 1024 * 1024; // 10MB

    public static void uploadToHDFS(String localFilePath, String hdfsFilePath) throws IOException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);

        Path localPath = new Path(localFilePath);
        Path hdfsPath = new Path(hdfsFilePath);

        fs.copyFromLocalFile(false, true, localPath, hdfsPath);

        fs.close();
    }

    public static void uploadToHBase(String localFilePath, String hbaseTableName, String hbaseRowKey) throws IOException {
        Configuration conf = new Configuration();
        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(Bytes.toBytes(hbaseTableName));

        File file = new File(localFilePath);
        byte[] fileBytes = Files.readAllBytes(file.toPath());

        Put put = new Put(Bytes.toBytes(hbaseRowKey));
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("file"), fileBytes);

        table.put(put);

        table.close();
        connection.close();
    }

    public static void main(String[] args) {
        String localFilePath = "path/to/local/file.txt";
        String hdfsFilePath = "hdfs:///path/to/hdfs/file.txt";
        String hbaseTableName = "table_name";
        String hbaseRowKey = "row_key";

        File file = new File(localFilePath);
        long fileSize = file.length();

        if (fileSize > MAX_HDFS_UPLOAD_SIZE) {
            try {
                uploadToHDFS(localFilePath, hdfsFilePath);
                System.out.println("File uploaded to HDFS");
            } catch (IOException e) {
                System.err.println("Failed to upload file to HDFS: " + e.getMessage());
            }
        } else {
            try {
                uploadToHBase(localFilePath, hbaseTableName, hbaseRowKey);
                System.out.println("File uploaded to HBase");
            } catch (IOException e) {
                System.err.println("Failed to upload file to HBase: " + e.getMessage());
            }
        }
    }
}

在上述代码中，首先定义了一个阈值 MAX_HDFS_UPLOAD_SIZE，设置为 10MB。然后，定义了两个方法 uploadToHDFS() 和 uploadToHBase()，分别用于向 HDFS 和 HBase 中上传文件。

在 main() 方法中，它会根据文件的大小选择上传方式。如果文件大小大于阈值，调用 uploadToHDFS() 方法将文件上传到 HDFS；否则，调用 uploadToHBase() 方法将文件上传到 HBase。

在测试时，将 localFilePath 修改为本地文件的路径，hdfsFilePath 修改为 HDFS 中保存的路径，hbaseTableName 修改为 HBase 表的名称，hbaseRowKey 修改为 HBase 行的唯一标识。

请确保您已经配置好相应的 HDFS 和 HBase 环境，并在项目中添加相应的依赖。

希望以上代码对您有所帮助！如有其他问题，请随时提问。