springboot集成hadoop实战

素颜马尾好姑娘i 2022-12-25 11:56 218阅读 0赞

springboot集成hadoop实现hdfs增删改查

## maven坐标 ##

<dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-common</artifactId>
                <version>${hadoop.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-streaming</artifactId>
                <version>${hadoop.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-yarn-common</artifactId>
                <version>${hadoop.version}</version>
                <exclusions>
                    <exclusion>
                        <groupId>com.google.guava</groupId>
                        <artifactId>guava</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-distcp</artifactId>
                <version>${hadoop.version}</version>
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-mapreduce-client-core</artifactId>
                <version>${hadoop.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-hdfs</artifactId>
                <version>${hadoop.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
                <version>${hadoop.version}</version>
                <scope>provided</scope>
            </dependency>
    
    				
            <dependency>
                <groupId>cn.bestwu</groupId>
                <artifactId>ik-analyzers</artifactId>
                <version>5.1.0</version>
            </dependency>

## 配置 ##

hdfs的配置

hdfs:
      hdfsPath: hdfs://bigdata-master:8020
      hdfsName: bigdata-master

将fileSystem配置并注册到spring容器

@Slf4j
    @Configuration
    public class HadoopHDFSConfiguration { 
    
        @Value("${hdfs.hdfsPath}")
        private String hdfsPath;
        @Value("${hdfs.hdfsName}")
        private String hdfsName;
    
        @Bean
        public org.apache.hadoop.conf.Configuration  getConfiguration(){ 
            org.apache.hadoop.conf.Configuration configuration = new org.apache.hadoop.conf.Configuration();
            configuration.set("fs.defaultFS", hdfsPath);
            return configuration;
        }
    
        @Bean
        public FileSystem getFileSystem(){ 
            FileSystem fileSystem = null;
            try { 
                fileSystem = FileSystem.get(new URI(hdfsPath), getConfiguration(), hdfsName);
            } catch (IOException e) { 
                // TODO Auto-generated catch block
                log.error(e.getMessage());
            } catch (InterruptedException e) { 
                // TODO Auto-generated catch block
                log.error(e.getMessage());
            } catch (URISyntaxException e) { 
                // TODO Auto-generated catch block
                log.error(e.getMessage());
            }
            return fileSystem;
        }
    
    }

## 增删改查 ##

public interface HDFSService { 
    
    		// 创建文件夹
        boolean makeFolder(String path);
    		// 是否存在文件
        boolean existFile(String path);
    		
        List<Map<String, Object>> readCatalog(String path);
    
        boolean createFile(String path, MultipartFile file);
    
        String readFileContent(String path);
    
        List<Map<String, Object>> listFile(String path);
    
        boolean renameFile(String oldName, String newName);
    
        boolean deleteFile(String path);
    
        boolean uploadFile(String path, String uploadPath);
    
        boolean downloadFile(String path, String downloadPath);
    
        boolean copyFile(String sourcePath, String targetPath);
    
        byte[] openFileToBytes(String path);
    
        BlockLocation[] getFileBlockLocations(String path);
    
    }

@Slf4j
    @Service
    public class HDFSServiceImpl implements HDFSService { 
    
        private static final int bufferSize = 1024 * 1024 * 64;
    
        @Autowired
        private FileSystem fileSystem;
    
        @Override
        public boolean makeFolder(String path) { 
            boolean target = false;
            if (StringUtils.isEmpty(path)) { 
                return false;
            }
            if (existFile(path)) { 
                return true;
            }
            Path src = new Path(path);
            try { 
                target = fileSystem.mkdirs(src);
            } catch (IOException e) { 
                log.error(e.getMessage());
            }
            return target;
        }
    
        @Override
        public boolean existFile(String path) { 
            if (StringUtils.isEmpty(path)){ 
                return false;
            }
            Path src = new Path(path);
            try { 
                return fileSystem.exists(src);
            } catch (IOException e) { 
                log.error(e.getMessage());
            }
            return false;
        }
    
        @Override
        public List<Map<String, Object>> readCatalog(String path) { 
            if (StringUtils.isEmpty(path)){ 
                return Collections.emptyList();
            }
            if (!existFile(path)){ 
                log.error("catalog is not exist!!");
                return Collections.emptyList();
            }
    
            Path src = new Path(path);
            FileStatus[] fileStatuses = null;
            try { 
                fileStatuses = fileSystem.listStatus(src);
            } catch (IOException e) { 
                log.error(e.getMessage());
            }
            List<Map<String, Object>> result = new ArrayList<>(fileStatuses.length);
    
            if (null != fileStatuses && 0 < fileStatuses.length) { 
                for (FileStatus fileStatus : fileStatuses) { 
                    Map<String, Object> cataLogMap = new HashMap<>();
                    cataLogMap.put("filePath", fileStatus.getPath());
                    cataLogMap.put("fileStatus", fileStatus);
                    result.add(cataLogMap);
                }
            }
            return result;
        }
    
        @Override
        public boolean createFile(String path, MultipartFile file) { 
            boolean target = false;
            if (StringUtils.isEmpty(path)) { 
                return false;
            }
            String fileName = file.getName();
            Path newPath = new Path(path + "/" + fileName);
    
            FSDataOutputStream outputStream = null;
            try { 
                outputStream = fileSystem.create(newPath);
                outputStream.write(file.getBytes());
                target = true;
            } catch (IOException e) { 
                log.error(e.getMessage());
            } finally { 
                if (null != outputStream) { 
                    try { 
                        outputStream.close();
                    } catch (IOException e) { 
                        log.error(e.getMessage());
                    }
                }
            }
            return target;
        }
    
        @Override
        public String readFileContent(String path) { 
            if (StringUtils.isEmpty(path)){ 
                return null;
            }
    
            if (!existFile(path)) { 
                return null;
            }
    
            Path src = new Path(path);
    
            FSDataInputStream inputStream = null;
            StringBuilder sb = new StringBuilder();
            try { 
                inputStream = fileSystem.open(src);
                String lineText = "";
                while ((lineText = inputStream.readLine()) != null) { 
                    sb.append(lineText);
                }
            } catch (IOException e) { 
                log.error(e.getMessage());
            } finally { 
                if (null != inputStream) { 
                    try { 
                        inputStream.close();
                    } catch (IOException e) { 
                        log.error(e.getMessage());
                    }
                }
            }
            return sb.toString();
        }
    
        @Override
        public List<Map<String, Object>> listFile(String path) { 
            if (StringUtils.isEmpty(path)) { 
                return Collections.emptyList();
            }
            if (!existFile(path)) { 
                return Collections.emptyList();
            }
            List<Map<String,Object>> resultList = new ArrayList<>();
    
            Path src = new Path(path);
            try { 
                RemoteIterator<LocatedFileStatus> fileIterator = fileSystem.listFiles(src, true);
                while (fileIterator.hasNext()) { 
                    LocatedFileStatus next = fileIterator.next();
                    Path filePath = next.getPath();
                    String fileName = filePath.getName();
                    Map<String, Object> map = new HashMap<>();
                    map.put("fileName", fileName);
                    map.put("filePath", filePath.toString());
                    resultList.add(map);
                }
            } catch (IOException e) { 
                log.error(e.getMessage());
            }
    
            return resultList;
        }
    
        @Override
        public boolean renameFile(String oldName, String newName) { 
            boolean target = false;
            if (StringUtils.isEmpty(oldName) || StringUtils.isEmpty(newName)) { 
                return false;
            }
            Path oldPath = new Path(oldName);
            Path newPath = new Path(newName);
            try { 
                target = fileSystem.rename(oldPath, newPath);
            } catch (IOException e) { 
                log.error(e.getMessage());
            }
    
            return target;
        }
    
        @Override
        public boolean deleteFile(String path) { 
            boolean target = false;
            if (StringUtils.isEmpty(path)) { 
                return false;
            }
            if (!existFile(path)) { 
                return false;
            }
            Path src = new Path(path);
            try { 
                target = fileSystem.deleteOnExit(src);
            } catch (IOException e) { 
                log.error(e.getMessage());
            }
            return target;
        }
    
        @Override
        public boolean uploadFile(String path, String uploadPath) { 
            if (StringUtils.isEmpty(path) || StringUtils.isEmpty(uploadPath)) { 
                return false;
            }
    
            Path clientPath = new Path(path);
    
            Path serverPath = new Path(uploadPath);
    
            try { 
                fileSystem.copyFromLocalFile(false,clientPath,serverPath);
                return true;
            } catch (IOException e) { 
                log.error(e.getMessage(), e);
            }
            return false;
        }
    
        @Override
        public boolean downloadFile(String path, String downloadPath) { 
            if (StringUtils.isEmpty(path) || StringUtils.isEmpty(downloadPath)) { 
                return false;
            }
    
            Path clienPath = new Path(path);
    
            Path targetPath = new Path(downloadPath);
    
            try { 
                fileSystem.copyToLocalFile(false,clienPath, targetPath);
                return true;
            } catch (IOException e) { 
                log.error(e.getMessage());
            }
            return false;
        }
    
        @Override
        public boolean copyFile(String sourcePath, String targetPath) { 
            if (StringUtils.isEmpty(sourcePath) || StringUtils.isEmpty(targetPath)) { 
                return false;
            }
    
            Path oldPath = new Path(sourcePath);
    
            Path newPath = new Path(targetPath);
    
            FSDataInputStream inputStream = null;
            FSDataOutputStream outputStream = null;
    
            try { 
                inputStream = fileSystem.open(oldPath);
                outputStream = fileSystem.create(newPath);
    
                IOUtils.copyBytes(inputStream,outputStream,bufferSize,false);
                return true;
            } catch (IOException e) { 
                log.error(e.getMessage());
            } finally { 
                if (null != inputStream) { 
                    try { 
                        inputStream.close();
                    } catch (IOException e) { 
                        log.error(e.getMessage());
                    }
                }
                if (null != outputStream) { 
                    try { 
                        outputStream.close();
                    } catch (IOException e) { 
                        log.error(e.getMessage());
                    }
                }
            }
            return false;
        }
    
        @Override
        public byte[] openFileToBytes(String path) { 
    
            if (StringUtils.isEmpty(path)) { 
                return null;
            }
    
            if (!existFile(path)) { 
                return null;
            }
    
            Path src = new Path(path);
            byte[] result = null;
            FSDataInputStream inputStream = null;
            try { 
                inputStream = fileSystem.open(src);
                result = IOUtils.readFullyToByteArray(inputStream);
            } catch (IOException e) { 
                log.error(e.getMessage());
            } finally { 
                if (null != inputStream){ 
                    try { 
                        inputStream.close();
                    } catch (IOException e) { 
                        log.error(e.getMessage());
                    }
                }
            }
    
            return result;
        }
    
        @Override
        public BlockLocation[] getFileBlockLocations(String path) { 
            if (StringUtils.isEmpty(path)) { 
                return null;
            }
            if (!existFile(path)) { 
                return null;
            }
            BlockLocation[] blocks = null;
            Path src = new Path(path);
            try{ 
                FileStatus fileStatus = fileSystem.getFileStatus(src);
                blocks = fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
            }catch(Exception e){ 
                log.error(e.getMessage());
            }
            return blocks;
        }
    }

## mapReduce ##

package com.winterchen.hadoopdemo.reduce;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    
    /* * 继承Reducer类需要定义四个输出、输出类型泛型： * 四个泛型类型分别代表： * KeyIn Reducer的输入数据的Key，这里是每行文字中的单词"hello" * ValueIn Reducer的输入数据的Value，这里是每行文字中的次数 * KeyOut Reducer的输出数据的Key，这里是每行文字中的单词"hello" * ValueOut Reducer的输出数据的Value，这里是每行文字中的出现的总次数 */
    public class WordReduce extends Reducer<Text, IntWritable, Text, IntWritable> { 
    
        private IntWritable result = new IntWritable();
        private List<String> textList = new ArrayList<>();
    
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { 
            int sum = 0;
            for (IntWritable val : values) { 
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
    
            String keyStr = key.toString();
    
            // 使用分词器，内容已经被统计好了，直接输出即可
            if (textList.contains(keyStr)) { 
                System.out.println("============ " + keyStr + " 统计分词为: " + sum + " ============");
            }
        }
    }

package com.winterchen.hadoopdemo.configuration;
    
    import com.winterchen.hadoopdemo.HadoopDemoApplication;
    import com.winterchen.hadoopdemo.mapper.WordMapper;
    import com.winterchen.hadoopdemo.reduce.WordReduce;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    import java.io.IOException;
    
    @Component
    public class ReduceJobsConfiguration { 
    
        @Value("${hdfs.hdfsPath}")
        private String hdfsPath;
    
        /** * 获取HDFS配置信息 * * @return */
        public Configuration getConfiguration() { 
            Configuration configuration = new Configuration();
            configuration.set("fs.defaultFS", hdfsPath);
            configuration.set("mapred.job.tracker", hdfsPath);
            return configuration;
        }
    
        /** * 获取单词统计的配置信息 * * @param jobName * @param inputPath * @param outputPath * @throws IOException * @throws ClassNotFoundException * @throws InterruptedException */
        public void getWordCountJobsConf(String jobName, String inputPath, String outputPath)
                throws IOException, ClassNotFoundException, InterruptedException { 
            Configuration conf = getConfiguration();
            Job job = Job.getInstance(conf, jobName);
    
            job.setMapperClass(WordMapper.class);
            job.setCombinerClass(WordReduce.class);
            job.setJarByClass(HadoopDemoApplication.class);
            job.setReducerClass(WordReduce.class);
    
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            FileInputFormat.addInputPath(job, new Path(inputPath));
            FileOutputFormat.setOutputPath(job, new Path(outputPath));
            job.waitForCompletion(true);
        }
    
        @PostConstruct
        public void getPath() { 
            hdfsPath = this.hdfsPath;
        }
    
        public String getHdfsPath() { 
            return hdfsPath;
        }
    }

public interface MapReduceService { 
    
        void wordCount(String jobName, String inputPath, String outputPath) throws Exception;
    
    }

package com.winterchen.hadoopdemo.service.impl;
    
    import com.winterchen.hadoopdemo.configuration.ReduceJobsConfiguration;
    import com.winterchen.hadoopdemo.service.HDFSService;
    import com.winterchen.hadoopdemo.service.MapReduceService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import org.springframework.util.StringUtils;
    
    @Service
    public class MapReduceServiceImpl implements MapReduceService { 
    
        @Autowired
        private HDFSService hdfsService;
    
        @Autowired
        private ReduceJobsConfiguration reduceJobsConfiguration;
    
        @Override
        public void wordCount(String jobName, String inputPath, String outputPath) throws Exception { 
            if (StringUtils.isEmpty(jobName) || StringUtils.isEmpty(inputPath)) { 
                return;
            }
            // 输出目录 = output/当前Job,如果输出路径存在则删除，保证每次都是最新的
            if (hdfsService.existFile(outputPath)) { 
                hdfsService.deleteFile(outputPath);
            }
            reduceJobsConfiguration.getWordCountJobsConf(jobName, inputPath, outputPath);
        }
    }

@Slf4j
    @Api(tags = "map reduce api")
    @RestController
    @RequestMapping("/api/v1/map-reduce")
    public class MapReduceController { 
    
        @Autowired
        private MapReduceService mapReduceService;
    
        @ApiOperation("count word")
        @PostMapping("/word/count")
        public APIResponse wordCount(
                @ApiParam(name = "jobName", required = true)
                @RequestParam(name = "jobName", required = true)
                String jobName,
                @ApiParam(name = "inputPath", required = true)
                @RequestParam(name = "inputPath", required = true)
                String inputPath,
                @ApiParam(name = "outputPath", required = true)
                @RequestParam(name = "outputPath", required = true)
                String outputPath
        ){ 
            try { 
                mapReduceService.wordCount(jobName, inputPath, outputPath);
                return APIResponse.success();
            } catch (Exception e) { 
                log.error(e.getMessage());
                return APIResponse.fail(e.getMessage());
            }
        }
    }

以上就是日常开发中能使用到的基本的功能：hdfs的增删改查，以及MapReduce；

源码地址:

[WinterChenS/springboot-learning-experience][WinterChenS_springboot-learning-experience]

[WinterChenS_springboot-learning-experience]: https://github.com/WinterChenS/springboot-learning-experience/tree/master/spring-boot-hadoop