SpringBoot集成Hadoop实现文件的上传和下载功能

        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>2.5.8</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.3.6</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.3.6</version>
        </dependency>
		<!--需要指定hadoop-auth、hadoop-common版本，否则容易出现版本冲突-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-auth</artifactId>
            <version>3.3.6</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>3.3.6</version>
        </dependency>

hadoop:
  name-node: hdfs://test:8020

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.FileSystem;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.net.URI;
/**
 * @description: Hadoop配置类
 * @version: 1.0
 */
@Configuration
@Slf4j
public class HadoopConfig {
    @Value("${hadoop.name-node}")
    private String nameNode;
    @Bean("fileSystem")
    public FileSystem fileSystem() {
        //读取配置文件
        org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
        conf.set("dfs.replication", "1");
        // 文件系统
        FileSystem fileSystem = null;
        // 返回指定的文件系统,如果在本地测试，需要使用此种方法获取文件系统
        try {
            URI uri = new URI(nameNode.trim());
            fileSystem = FileSystem.get(new URI(nameNode), conf, "root");
        } catch (Exception e) {
            log.error("文件系统获取发生异常", e);
        }
        return fileSystem;
    }
}

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.stereotype.Component;
import java.io.*;
/**
 * @description: hadoop操作类
 * @version: 1.0
 */
@Slf4j
@Component
public class HadoopUtil {
    @Resource
    private FileSystem fileSystem;
    /**
     * 上传文件
     * @param file 文件
     * @param destPath 上传的路径
     */
    public void uploadFile(File file, String destPath) {
        try (InputStream inputStream = new FileInputStream(file);
             FSDataOutputStream outputStream = fileSystem.create(new Path(destPath))) {
            byte[] buffer = new byte[1024];
            int bytesRead;
            while ((bytesRead = inputStream.read(buffer)) != -1) {
                outputStream.write(buffer, 0, bytesRead);
            }
            // 刷新输出流以确保所有数据都被写入HDFS
            outputStream.flush();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    /**
     * 下载文件
     * @param filePath 文件路径
     * @return InputStream 返回inputStream类型  根据自己的需求就行转换
     * @throws IOException
     */
    public InputStream download(String filePath) throws IOException {
        return fileSystem.open(new Path(filePath));
    }
    /**
     * 删除文件
     * @param filePath 文件路径
     * @throws IOException
     */
    public void deleteFile(String filePath) throws IOException {
        fileSystem.delete(new Path(filePath), true);
    }
}

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import com.xxx.xxx.HadoopUtil;
import java.io.File;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
@Slf4j
@Service
public class Test {
	@Resource
    private HadoopUtil hadoopUtil;
	public String test(){
		// 一、本地文件上传hdfs
        File file = new File("/you/file/path/文件名");
        hadoopUtil.uploadFile(file, "/you/upload/path/文件名");
		// 二、下载hadoop中的文件并传为String
        try (InputStream in = hadoopUtil.download("/文件名")) {
            String str = new Scanner(in, StandardCharsets.UTF_8).useDelimiter("\\A").next();
            log.info("结果" + str);
        } catch (IOException e) {
            log.error("文件导出发生异常")
        }
    }
}

[root@test app]# hdfs dfs -ls /cda
Found 1 items
-rw-r--r--   1 root supergroup      23005 2024-07-08 17:13 /cda/EMR-SD-01-病历概要-T01-001.xml