SpringBoot2.* Hadoop3.3

前言：HDFS安装参照

CentOS7搭建HDFS（Hadoop3.3）

1、在SpringBoot项目中添加Hadoop客户端操作的pom依赖

:::tips Hadoop核心依赖包采用log4j方式记录日志，spring-boot-start-web 核心jar包日志采用logback 方式记录。这里采用spring-boot-start-web的日志记录方式而排除log4j的相关依赖 :::

  1. <dependency>
  2.   <groupId>org.apache.hadoop</groupId>
  3.   <artifactId>hadoop-common</artifactId>
  4.   <version>3.3.0</version>
  5.   <exclusions>
  6.     <exclusion>
  7.       <groupId>org.slf4j</groupId>
  8.       <artifactId>slf4j-log4j12</artifactId>
  9.     </exclusion>
  10.   </exclusions>
  11. </dependency>
  12. <dependency>
  13.   <groupId>org.apache.hadoop</groupId>
  14.   <artifactId>hadoop-hdfs</artifactId>
  15.   <version>3.3.0</version>
  16.   <exclusions>
  17.     <exclusion>
  18.       <groupId>org.slf4j</groupId>
  19.       <artifactId>slf4j-log4j12</artifactId>
  20.     </exclusion>
  21.   </exclusions>
  22. </dependency>
  23. <dependency>
  24.   <groupId>org.apache.hadoop</groupId>
  25.   <artifactId>hadoop-client</artifactId>
  26.   <version>3.3.0</version>
  27.   <exclusions>
  28.     <exclusion>
  29.       <groupId>org.slf4j</groupId>
  30.       <artifactId>slf4j-log4j12</artifactId>
  31.     </exclusion>
  32.   </exclusions>
  33. </dependency>
  34. <dependency>
  35.   <groupId>com.alibaba</groupId>
  36.   <artifactId>fastjson</artifactId>
  37.   <version>1.2.72</version>
  38. </dependency>

2、Hadoop Client需要做的配置

在客户端环境变量配置HADOOP_USER_NAME

因为在启动Hadoop时，使用root用户启动，Hadoop客户端会读取环境变量中配置的用户名，如果没有则默认会读取系统计算机管理员名称，会报无权限的异常。
image.png

如果是访问云主机，则需要在本地配置主机名称和公网IP的映射

image.png

3、在Java中配置HDFS的信息

A.在ymal配置文件配置主机信息

  1. application:
  2.   config:
  3.     hdfs:
  4.       host: hdfs://120.79.178.68:9100

B.在Config配置文件引用配置参数即可

:::info 以下几个配置项说明：

  1. dfs.replication：设置副本的数量，在JAVA API客户端设置该参数只是为了和服务端启动时设置的副本数量保持一致，不设置则默认为3个副本。
  2. dfs.client.use.datanode.hostname：解决服务Java 操作HDFS时，NameNode 和DataNode 是通过局域网进行通信，NameNode返回地址为 DataNode 的私有 IP，本地无法访问的问题。 :::

    问题的原因：NameNode节点存放的是文件目录，也就是文件夹、文件名称，本地可以通过公网访问 NameNode，所以可以进行文件夹的创建，当上传文件需要写入数据到DataNode时，NameNode 和DataNode 是通过局域网进行通信，NameNode返回地址为 DataNode 的私有 IP，本地无法访问。

  1. import io.renren.service.HdfsService;
  2. import org.springframework.context.annotation.Bean;
  3. import org.springframework.context.annotation.Configuration;
  5. /**
  6.  * HDFS相关配置
  7.  *
  8.  * @author Fcant
  9.  * @since 1.0.0
  10.  */
  11. @Configuration
  12. public class HdfsConfig {
  13.     @Value("${application.config.hdfs.host}")
  14.     private String defaultHdfsUri;
  16.     @Bean
  17.     public HdfsService getHbaseService(){
  18.         org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
  19.         conf.set("fs.defaultFS",defaultHdfsUri);
  20.         conf.set("dfs.client.use.datanode.hostname", "true");
  21.         conf.set("dfs.replication", "1");
  22.         return new HdfsService(conf,defaultHdfsUri);
  23.     }
  24. }

4、通用Java操作HDFS的Service

  1. import com.alibaba.fastjson.JSON;
  2. import org.apache.commons.io.IOUtils;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.BlockLocation;
  5. import org.apache.hadoop.fs.FSDataInputStream;
  6. import org.apache.hadoop.fs.FileStatus;
  7. import org.apache.hadoop.fs.FileSystem;
  8. import org.apache.hadoop.fs.Path;
  9. import org.apache.hadoop.fs.PathFilter;
  10. import org.slf4j.Logger;
  11. import org.slf4j.LoggerFactory;
  13. import java.io.IOException;
  14. import java.nio.charset.Charset;
  15. import java.text.MessageFormat;
  16. import java.util.ArrayList;
  17. import java.util.HashMap;
  18. import java.util.List;
  19. import java.util.Map;
  21. /**
  22.  * HDFS相关的基本操作
  23.  *
  24.  * @author Fcant
  25.  * @since 1.0.0
  26.  */
  27. public class HdfsService {
  29.     private Logger logger = LoggerFactory.getLogger(HdfsService.class);
  30.     private Configuration conf = null;
  32.     /**
  33.      * 默认的HDFS路径，比如：hdfs://192.168.197.130:9000
  34.      */
  35.     private String defaultHdfsUri;
  37.     public HdfsService(Configuration conf,String defaultHdfsUri) {
  38.         this.conf = conf;
  39.         this.defaultHdfsUri = defaultHdfsUri;
  40.     }
  42.     /**
  43.      * 获取HDFS文件系统
  44.      * @return org.apache.hadoop.fs.FileSystem
  45.      */
  46.     private FileSystem getFileSystem() throws IOException {
  47.         return FileSystem.get(conf);
  48.     }
  50.     /**
  51.      * 创建HDFS目录
  52.      * @author Fcant
  53.      * @since 1.0.0
  54.      * @param path HDFS的相对目录路径，比如：/testDir
  55.      * @return boolean 是否创建成功
  56.      */
  57.     public boolean mkdir(String path){
  58.         //如果目录已经存在，则直接返回
  59.         if(checkExists(path)){
  60.             return true;
  61.         }else{
  62.             FileSystem fileSystem = null;
  64.             try {
  65.                 fileSystem = getFileSystem();
  67.                 //最终的HDFS文件目录
  68.                 String hdfsPath = generateHdfsPath(path);
  69.                 //创建目录
  70.                 return fileSystem.mkdirs(new Path(hdfsPath));
  71.             } catch (IOException e) {
  72.                 logger.error(MessageFormat.format("创建HDFS目录失败，path:{0}",path),e);
  73.                 return false;
  74.             }finally {
  75.                 close(fileSystem);
  76.             }
  77.         }
  78.     }
  80.     /**
  81.      * 接口上传的文件上传到HDFS
  82.      *
  83.      * @param filePath 文件路径及文件名 Example：/business/2020/12/23/fbfc71d2-7/hello.png
  84.      * @param file 接口收到的文件
  85.      * @author Fcant 上午 10:36 2020/12/23/0023
  86.      */
  87.     public void uploadFileToHdfs(String filePath, MultipartFile file) {
  88.         Path path = new Path(filePath);
  89.         FSDataOutputStream outputStream;
  90.         try {
  91.             outputStream = getFileSystem().create(path);
  92.             outputStream.write(file.getBytes());
  93.             outputStream.close();
  94.         } catch (IOException e) {
  95.             e.printStackTrace();
  96.         }
  97.     }
  99.     /**
  100.      * 上传文件至HDFS
  101.      * @author Fcant
  102.      * @since 1.0.0
  103.      * @param srcFile 本地文件路径，比如：D:/test.txt
  104.      * @param dstPath HDFS的相对目录路径，比如：/testDir
  105.      */
  106.     public void uploadFileToHdfs(String srcFile, String dstPath){
  107.         this.uploadFileToHdfs(false, true, srcFile, dstPath);
  108.     }
  110.     /**
  111.      * 上传文件至HDFS
  112.      * @author Fcant
  113.      * @since 1.0.0
  114.      * @param delSrc 是否删除本地文件
  115.      * @param overwrite 是否覆盖HDFS上面的文件
  116.      * @param srcFile 本地文件路径，比如：D:/test.txt
  117.      * @param dstPath HDFS的相对目录路径，比如：/testDir
  118.      */
  119.     public void uploadFileToHdfs(boolean delSrc, boolean overwrite, String srcFile, String dstPath){
  120.         //源文件路径
  121.         Path localSrcPath = new Path(srcFile);
  122.         //目标文件路径
  123.         Path hdfsDstPath = new Path(generateHdfsPath(dstPath));
  125.         FileSystem fileSystem = null;
  126.         try {
  127.             fileSystem = getFileSystem();
  129.             fileSystem.copyFromLocalFile(delSrc,overwrite,localSrcPath,hdfsDstPath);
  130.         } catch (IOException e) {
  131.             logger.error(MessageFormat.format("上传文件至HDFS失败，srcFile:{0},dstPath:{1}",srcFile,dstPath),e);
  132.         }finally {
  133.             close(fileSystem);
  134.         }
  135.     }
  137.     /**
  138.      * 判断文件或者目录是否在HDFS上面存在
  139.      * @author Fcant
  140.      * @since 1.0.0
  141.      * @param path HDFS的相对目录路径，比如：/testDir、/testDir/a.txt
  142.      * @return boolean
  143.      */
  144.     public boolean checkExists(String path){
  145.         FileSystem fileSystem = null;
  146.         try {
  147.             fileSystem = getFileSystem();
  149.             //最终的HDFS文件目录
  150.             String hdfsPath = generateHdfsPath(path);
  152.             //创建目录
  153.             return fileSystem.exists(new Path(hdfsPath));
  154.         } catch (IOException e) {
  155.             logger.error(MessageFormat.format("'判断文件或者目录是否在HDFS上面存在'失败，path:{0}",path),e);
  156.             return false;
  157.         }finally {
  158.             close(fileSystem);
  159.         }
  160.     }
  162.     /**
  163.      * 获取HDFS上面的某个路径下面的所有文件或目录（不包含子目录）信息
  164.      * @author Fcant
  165.      * @since 1.0.0
  166.      * @param path HDFS的相对目录路径，比如：/testDir
  167.      * @return java.util.List<java.util.Map<java.lang.String,java.lang.Object>>
  168.      */
  169.     public List<Map<String,Object>> listFiles(String path, PathFilter pathFilter){
  170.         //返回数据
  171.         List<Map<String,Object>> result = new ArrayList<>();
  173.         //如果目录已经存在，则继续操作
  174.         if(checkExists(path)){
  175.             FileSystem fileSystem = null;
  177.             try {
  178.                 fileSystem = getFileSystem();
  180.                 //最终的HDFS文件目录
  181.                 String hdfsPath = generateHdfsPath(path);
  183.                 FileStatus[] statuses;
  184.                 //根据Path过滤器查询
  185.                 if(pathFilter != null){
  186.                     statuses = fileSystem.listStatus(new Path(hdfsPath),pathFilter);
  187.                 }else{
  188.                     statuses = fileSystem.listStatus(new Path(hdfsPath));
  189.                 }
  191.                 if(statuses != null){
  192.                     for(FileStatus status : statuses){
  193.                         //每个文件的属性
  194.                         Map<String,Object> fileMap = new HashMap<>(2);
  196.                         fileMap.put("path",status.getPath().toString());
  197.                         fileMap.put("isDir",status.isDirectory());
  199.                         result.add(fileMap);
  200.                     }
  201.                 }
  202.             } catch (IOException e) {
  203.                 logger.error(MessageFormat.format("获取HDFS上面的某个路径下面的所有文件失败，path:{0}",path),e);
  204.             }finally {
  205.                 close(fileSystem);
  206.             }
  207.         }
  209.         return result;
  210.     }
  213.     /**
  214.      * 从HDFS下载文件至本地
  215.      * @author Fcant
  216.      * @since 1.0.0
  217.      * @param srcFile HDFS的相对目录路径，比如：/testDir/a.txt
  218.      * @param dstFile 下载之后本地文件路径（如果本地文件目录不存在，则会自动创建），比如：D:/test.txt
  219.      */
  220.     public void downloadFileFromHdfs(String srcFile, String dstFile){
  221.         //HDFS文件路径
  222.         Path hdfsSrcPath = new Path(generateHdfsPath(srcFile));
  223.         //下载之后本地文件路径
  224.         Path localDstPath = new Path(dstFile);
  226.         FileSystem fileSystem = null;
  227.         try {
  228.             fileSystem = getFileSystem();
  230.             fileSystem.copyToLocalFile(hdfsSrcPath,localDstPath);
  231.         } catch (IOException e) {
  232.             logger.error(MessageFormat.format("从HDFS下载文件至本地失败，srcFile:{0},dstFile:{1}",srcFile,dstFile),e);
  233.         }finally {
  234.             close(fileSystem);
  235.         }
  236.     }
  238.     /**
  239.      * 打开HDFS上面的文件并返回 InputStream
  240.      * @author Fcant
  241.      * @since 1.0.0
  242.      * @param path HDFS的相对目录路径，比如：/testDir/c.txt
  243.      * @return FSDataInputStream
  244.      */
  245.     public FSDataInputStream open(String path){
  246.         //HDFS文件路径
  247.         Path hdfsPath = new Path(generateHdfsPath(path));
  249.         FileSystem fileSystem = null;
  250.         try {
  251.             fileSystem = getFileSystem();
  253.             return fileSystem.open(hdfsPath);
  254.         } catch (IOException e) {
  255.             logger.error(MessageFormat.format("打开HDFS上面的文件失败，path:{0}",path),e);
  256.         }
  258.         return null;
  259.     }
  261.     /**
  262.      * 打开HDFS上面的文件并返回byte数组，方便Web端下载文件
  263.      * <p>new ResponseEntity<byte[]>(byte数组, headers, HttpStatus.CREATED);</p>
  264.      * <p>或者：new ResponseEntity<byte[]>(FileUtils.readFileToByteArray(templateFile), headers, HttpStatus.CREATED);</p>
  265.      * @author Fcant
  266.      * @since 1.0.0
  267.      * @param path HDFS的相对目录路径，比如：/testDir/b.txt
  268.      * @return FSDataInputStream
  269.      */
  270.     public byte[] openWithBytes(String path){
  271.         //HDFS文件路径
  272.         Path hdfsPath = new Path(generateHdfsPath(path));
  274.         FileSystem fileSystem = null;
  275.         FSDataInputStream inputStream = null;
  276.         try {
  277.             fileSystem = getFileSystem();
  278.             inputStream = fileSystem.open(hdfsPath);
  280.             return IOUtils.toByteArray(inputStream);
  281.         } catch (IOException e) {
  282.             logger.error(MessageFormat.format("打开HDFS上面的文件失败，path:{0}",path),e);
  283.         }finally {
  284.             if(inputStream != null){
  285.                 try {
  286.                     inputStream.close();
  287.                 } catch (IOException e) {
  288.                     // ignore
  289.                 }
  290.             }
  291.         }
  293.         return null;
  294.     }
  296.     /**
  297.      * 打开HDFS上面的文件并返回String字符串
  298.      * @author Fcant
  299.      * @since 1.0.0
  300.      * @param path HDFS的相对目录路径，比如：/testDir/b.txt
  301.      * @return FSDataInputStream
  302.      */
  303.     public String openWithString(String path){
  304.         //HDFS文件路径
  305.         Path hdfsPath = new Path(generateHdfsPath(path));
  307.         FileSystem fileSystem = null;
  308.         FSDataInputStream inputStream = null;
  309.         try {
  310.             fileSystem = getFileSystem();
  311.             inputStream = fileSystem.open(hdfsPath);
  313.             return IOUtils.toString(inputStream, Charset.forName("UTF-8"));
  314.         } catch (IOException e) {
  315.             logger.error(MessageFormat.format("打开HDFS上面的文件失败，path:{0}",path),e);
  316.         }finally {
  317.             if(inputStream != null){
  318.                 try {
  319.                     inputStream.close();
  320.                 } catch (IOException e) {
  321.                     // ignore
  322.                 }
  323.             }
  324.         }
  326.         return null;
  327.     }
  329.     /**
  330.      * 打开HDFS上面的文件并转换为Java对象（需要HDFS上门的文件内容为JSON字符串）
  331.      * @author Fcant
  332.      * @since 1.0.0
  333.      * @param path HDFS的相对目录路径，比如：/testDir/c.txt
  334.      * @return FSDataInputStream
  335.      */
  336.     public <T extends Object> T openWithObject(String path, Class<T> clazz){
  337.         //1、获得文件的json字符串
  338.         String jsonStr = this.openWithString(path);
  340.         //2、使用com.alibaba.fastjson.JSON将json字符串转化为Java对象并返回
  341.         return JSON.parseObject(jsonStr, clazz);
  342.     }
  344.     /**
  345.      * 重命名
  346.      * @author Fcant
  347.      * @since 1.0.0
  348.      * @param srcFile 重命名之前的HDFS的相对目录路径，比如：/testDir/b.txt
  349.      * @param dstFile 重命名之后的HDFS的相对目录路径，比如：/testDir/b_new.txt
  350.      */
  351.     public boolean rename(String srcFile, String dstFile) {
  352.         //HDFS文件路径
  353.         Path srcFilePath = new Path(generateHdfsPath(srcFile));
  354.         //下载之后本地文件路径
  355.         Path dstFilePath = new Path(dstFile);
  357.         FileSystem fileSystem = null;
  358.         try {
  359.             fileSystem = getFileSystem();
  361.             return fileSystem.rename(srcFilePath,dstFilePath);
  362.         } catch (IOException e) {
  363.             logger.error(MessageFormat.format("重命名失败，srcFile:{0},dstFile:{1}",srcFile,dstFile),e);
  364.         }finally {
  365.             close(fileSystem);
  366.         }
  368.         return false;
  369.     }
  371.     /**
  372.      * 删除HDFS文件或目录
  373.      * @author Fcant
  374.      * @since 1.0.0
  375.      * @param path HDFS的相对目录路径，比如：/testDir/c.txt
  376.      * @return boolean
  377.      */
  378.     public boolean delete(String path) {
  379.         //HDFS文件路径
  380.         Path hdfsPath = new Path(generateHdfsPath(path));
  382.         FileSystem fileSystem = null;
  383.         try {
  384.             fileSystem = getFileSystem();
  386.             return fileSystem.delete(hdfsPath,true);
  387.         } catch (IOException e) {
  388.             logger.error(MessageFormat.format("删除HDFS文件或目录失败，path:{0}",path),e);
  389.         }finally {
  390.             close(fileSystem);
  391.         }
  393.         return false;
  394.     }
  396.     /**
  397.      * 获取某个文件在HDFS集群的位置
  398.      * @author Fcant
  399.      * @since 1.0.0
  400.      * @param path HDFS的相对目录路径，比如：/testDir/a.txt
  401.      * @return org.apache.hadoop.fs.BlockLocation[]
  402.      */
  403.     public BlockLocation[] getFileBlockLocations(String path) {
  404.         //HDFS文件路径
  405.         Path hdfsPath = new Path(generateHdfsPath(path));
  407.         FileSystem fileSystem = null;
  408.         try {
  409.             fileSystem = getFileSystem();
  410.             FileStatus fileStatus = fileSystem.getFileStatus(hdfsPath);
  412.             return fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
  413.         } catch (IOException e) {
  414.             logger.error(MessageFormat.format("获取某个文件在HDFS集群的位置失败，path:{0}",path),e);
  415.         }finally {
  416.             close(fileSystem);
  417.         }
  419.         return null;
  420.     }
  423.     /**
  424.      * 将相对路径转化为HDFS文件路径
  425.      * @author Fcant
  426.      * @since 1.0.0
  427.      * @param dstPath 相对路径，比如：/data
  428.      * @return java.lang.String
  429.      */
  430.     private String generateHdfsPath(String dstPath){
  431.         String hdfsPath = defaultHdfsUri;
  432.         if(dstPath.startsWith("/")){
  433.             hdfsPath += dstPath;
  434.         }else{
  435.             hdfsPath = hdfsPath + "/" + dstPath;
  436.         }
  438.         return hdfsPath;
  439.     }
  441.     /**
  442.      * close方法
  443.      */
  444.     private void close(FileSystem fileSystem){
  445.         if(fileSystem != null){
  446.             try {
  447.                 fileSystem.close();
  448.             } catch (IOException e) {
  449.                 logger.error(e.getMessage());
  450.             }
  451.         }
  452.     }
  453. }