AI-Backend/java_compile/src/main/java/cn/seecoder/ai/dao/HdfsDAO.java

package cn.seecoder.ai.dao;

import cn.seecoder.ai.config.AppConfig;
import cn.seecoder.ai.util.FileHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * this file is used to complete hdfs operates and mongodb operates
 * MongoDB, we now create user, fileLoc as collections
 */
@Component
@Slf4j
public class HdfsDAO {

    private final AppConfig appConfig;
    //connect to hdfs
    private FileSystem hdfs;
    private Configuration conf;

    private String hdfsServerAddr;
    private String hdfsUri;  //get hdfs Address


    @Autowired
    public HdfsDAO(AppConfig appConfig) {   //在初始化方法中使用到的bean需要以注入初始化方法的形式导入，否则读取到的引用可能是null。
        this.appConfig = appConfig;
        //hdfs主机地址 hdfs://ip:port/
        hdfsServerAddr= appConfig.getHdfsServerAddr();
        //hdfs存储位置 hdfs://ip:port/user-space/*/
        hdfsUri= appConfig.getHdfsUri();
        conf = new Configuration();

        setUserSpace();
        setConf();
    }

    //保证hdfs存储位置可以访问
    private void setUserSpace(){
        //先根据主机ip设置配置
        conf.set("fs.defaultFS", hdfsServerAddr);
        try {
            hdfs = FileSystem.get(conf);
            //若不存在文件夹，新建
            //若文件夹已存在，新建失败
            hdfs.mkdirs(new Path(hdfsUri));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //set the configuration and get connection
    private void setConf(){
        conf.set("fs.defaultFS", hdfsUri);
        log.info("当前主机下的hdfs目录是："+hdfsUri);
        //System.out.println(conf);
        try {
            hdfs = FileSystem.get(conf);
        } catch (Exception e) {
           e.printStackTrace();
        }

    }

    //put the local file to the hdfs
    public String putFile(String localFile){
        setConf();
        String fileName = FileHelper.getFile(localFile);//get the fileName
        String hdfsPathStr = hdfsUri + fileName;
        Path localPath = new Path(localFile);
        Path hdfsPath = new Path(hdfsPathStr);
        /*System.out.println(localPath);
        System.out.println(hdfsPath);*/
        try {
            hdfs.copyFromLocalFile(localPath,hdfsPath);
            hdfs.close();
            return hdfsPathStr;
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    //delete file not in mongodb
    public boolean deleteFileInHdfs(String path, boolean isFolder){
        setConf();
        log.info("delete HDFS file: " + path);
        boolean isdelete = false;
        Path hdfsPath = new Path(path);
        try {
            if (hdfs.exists(hdfsPath)) {
                isdelete = hdfs.delete(hdfsPath, isFolder);
            }
            hdfs.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        System.out.println(isdelete);
        return isdelete;
    }

    //get all files in data
    public FileStatus[] getStatus(String path){
        FileStatus[] fileStatuses =null;
        try {
            fileStatuses = hdfs.listStatus(new Path(path));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return  fileStatuses;
    }
    //
    public boolean exist(String path){
        setConf();
        log.info("file exist: " + path);
        Path localPath = new Path(path);
        try {
            if (hdfs.exists(localPath)) {
                hdfs.close();
                return true;
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return false;
    }
}