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; } }