hadoop Java API

package com.xhao.hadoop.util; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.*; import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.util.Progressable; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RestController; import java.io.*; @RestController public class HadoopUtils { @Autowired private FileSystem fileSystem; /** * 创建文件夹 * * @param path * @return * @throws Exception */ @PostMapping("mkdir") public boolean mkdir(String path) throws Exception { boolean flag = fileSystem.mkdirs(new Path(path)); return flag; } /** * 新建文件，遇到同名的直接覆盖 * * @param path * @param content * @throws Exception */ @PostMapping("create") public void create(String path, String content) throws Exception { FSDataOutputStream out = fileSystem.create(new Path(path), new Progressable() { public void progress() { System.out.println('+'); } }); out.write(content.getBytes()); out.flush(); out.close(); } @PostMapping("cat") public void cat() throws Exception { FSDataInputStream in = fileSystem.open(new Path("/testJavaApi/hello.txt")); IOUtils.copyBytes(in, System.out, 1024); } /** * @param oldName 全路径 /test/olename.txt * @param newName * @throws Exception */ @PostMapping("rename") public void rename(String oldName, String newName) throws Exception { Path oldPath = new Path(oldName); Path newPath = new Path(newName); boolean isSuccess = fileSystem.rename(oldPath, newPath); System.out.println(isSuccess); } /* * 上传文件到hdfs */ @PostMapping public void copyFromLocal(String source, String dest) throws Exception { Path localPath = new Path(source); Path hdfsPath = new Path(dest); fileSystem.copyFromLocalFile(localPath, hdfsPath); } /* * 上传文件到hdfs,输入输出流 */ @PostMapping("copyFromLocalwithPro") public void copyFromLocalwithPro(String filePath) throws Exception { Path hdfsPath = new Path("/testJavaApi/tongyi.exe"); InputStream in = new BufferedInputStream( new FileInputStream(new File("D:\\EasyConnectInstaller.exe"))); FSDataOutputStream out = fileSystem.create(hdfsPath, new Progressable() { public void progress() { System.out.println("="); } }); IOUtils.copyBytes(in, out, 2048); IOUtils.closeStream(out); IOUtils.closeStream(in); } /* * 下载文件到本地 */ public void copyToLocalFile() throws Exception { Path localPath = new Path("D:\\hello1.txt"); Path hdfsPath = new Path("/testJavaApi/hello.txt"); fileSystem.copyToLocalFile(hdfsPath, localPath); } public void delete() throws Exception { Path f = new Path("/testJavaApi/aaa.txt"); fileSystem.delete(f, true); } public void listFile() throws Exception { Path f = new Path("/testJavaApi"); FileStatus[] files = fileSystem.listStatus(f); for (FileStatus file : files) { String isDir = file.isDirectory() ? "文件夹" : "文件"; short rep = file.getReplication(); long len = file.getLen(); Path path = file.getPath(); System.out.println(isDir + "\t" + rep + "\t" + len + "\t" + path); } } //下载第一块block，只拷贝128M public void getBlock0() throws Exception, Exception { //获取输入流 FSDataInputStream fin = fileSystem.open(new Path("/hadoop-2.7.7.tar.gz")); //获取输出流 FileOutputStream fout = new FileOutputStream(new File("e:/hadoop.tar.gz")); //循环读取128M byte[] buf = new byte[1024]; for (int i = 0; i < 1024 * 128; i++) { fin.read(buf); fout.write(buf); } fout.close(); fin.close(); } //下载第二块block public void getBlock1() throws Exception, Exception { //获取输入流 FSDataInputStream fin = fileSystem.open(new Path("/hadoop-2.7.7.tar.gz")); //获取输出流 FileOutputStream fout = new FileOutputStream(new File("e:/hadoop.tar.gz.part1")); //指定读取的起点 fin.seek(1024 * 1024 * 128); IOUtils.copyBytes(fin, fout, new Configuration()); fout.close(); fin.close(); } }