【Hadoop】熟悉常用的HDFS操作（Java实现）

向右看齐 2023-07-13 12:16 8阅读 0赞

> # 一、实验目的 #
> 
> 1. 理解HDFS在Hadoop体系结构中的角色；
> 
> 2. 熟练使用HDFS操作常用的Shell命令；
> 
> 3. 熟悉HDFS操作常用的Java API。

> # 二、实验平台 #
> 
> 操作系统：Linux（deepin）
> 
> Hadoop版本：2.7.7
> 
> JDK版本：1.8
> 
> Java IDE：Eclipse

# 三、实验内容 #

### **1. 在分布式文件系统HDFS主目录，使用java 编程分别实现：** ###

1.  查看文件hadoop.txt是否存在，如果不存在，请创建。**（HdfsExpTask1）**
2.  hadoop.txt写入内容。**（HdfsExpTask2）**
3.  读取hadoop.txt文件内容。**（HdfsExpTask3）**

import java.io.BufferedReader;
    import java.io.FileInputStream;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.net.URI;
    import java.net.URISyntaxException;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FSDataInputStream;
    import org.apache.hadoop.fs.FSDataOutputStream;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    
    
    public class HdfsExpTask1 {
    
    	public static void main(String[] args) {
    		try {
    			Path filePath = new Path("hadoop.txt");
    			Configuration config = new Configuration();
    			config.set("fs.defaultFS", "hdfs://localhost:9000/");
    			FileSystem fs = FileSystem.get(config);
    			if(fs.exists(filePath)) {
    				System.out.println(filePath+" exists.");
    			} else {
    				FSDataOutputStream ops = fs.create(filePath);
    				ops.close();
    				fs.close();
    			}
    		} catch(Exception e) {
    			e.printStackTrace();
    		}
    
    	}
    
    }

import java.io.BufferedReader;
    import java.io.FileInputStream;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.net.URI;
    import java.net.URISyntaxException;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FSDataInputStream;
    import org.apache.hadoop.fs.FSDataOutputStream;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    
    public class HdfsExpTask2 {
    
    	public static void main(String[] args) {
    		try {
    			Path filePath = new Path("hadoop.txt");
    			Configuration config = new Configuration();
    			config.set("fs.defaultFs", "hdfs://localhost:9000/");
    			FileSystem fs = FileSystem.get(config);
    			String content = "Task2: WRITE(201701060413)";
    			byte[] buff = content.getBytes();
    			FSDataOutputStream ops = fs.create(filePath);
    			ops.write(buff, 0, buff.length);
    			ops.close();
    			fs.close();
    		}catch(Exception e) {
    			e.printStackTrace();
    		}
    	}
    
    }

import java.io.BufferedReader;
    
    public class HdfsExpTask3 {
    
    	public static void main(String[] args) {
    		try {
    			Path filePath = new Path("hadoop.txt");
    			Configuration config = new Configuration();
    			config.set("fs.defaultFS", "hdfs://localhost:9000/");
    			FileSystem fs = FileSystem.get(config);
    			
    			FSDataInputStream ips = fs.open(filePath);
    			BufferedReader reader = new BufferedReader(new InputStreamReader(ips));
    			String content = reader.readLine();
    			while(content != null) {
    				System.out.println(content);
    				content = reader.readLine();
    			}
    			reader.close();
    			ips.close();
    			fs.close();
    		}catch(Exception e) {
    			e.printStackTrace();
    		}
    
    	}
    
    }

### **2. Java编程实现以下指定功能：** ###

1.  将HDFS中指定文件的内容输出到终端中；
2.  给定HDFS中某一个目录，输出该目录下的所有文件的读写权限、大小、创建时间、路径等信息，如果该文件是目录，则递归输出该目录下所有文件相关信息；
3.  提供一个HDFS内的文件的路径，对该文件进行创建和删除操作。如果文件所在目录不存在，则自动创建目录；
4.  向HDFS中指定的文件追加内容，由用户指定内容追加到原有文件的开头或结尾；
5.  删除HDFS中指定的文件；
6.  删除HDFS中指定的目录，由用户指定目录中如果存在文件时是否删除目录；
7.  在HDFS中，将文件从源路径移动到目的路径。

import java.io.BufferedInputStream;
    import java.io.BufferedReader;
    import java.io.FileInputStream;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.net.URI;
    import java.net.URISyntaxException;
    import java.util.Scanner;
    
    import org.apache.commons.compress.utils.IOUtils;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FSDataInputStream;
    import org.apache.hadoop.fs.FSDataOutputStream;
    import org.apache.hadoop.fs.FileStatus;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.LocatedFileStatus;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.fs.RemoteIterator;
    
    public class HdfsExpTask4 {
    	/*
    	 *2.1 将HDFS中指定文件的内容输出到终端中 
    	 */
    	public static void cat(Configuration conf, String path) throws IOException{
    		FileSystem fs = FileSystem.get(conf);
    		Path filePath = new Path(path);
    		if(fs.exists(filePath)) {
    			FSDataInputStream in = fs.open(filePath);
    			BufferedReader br = new BufferedReader(new InputStreamReader(in));
    			String content = null;
    			while((content = br.readLine()) != null) {
    				System.out.println(content);
    			}
    			br.close();
    			fs.close();
    		}else {
    			System.out.println("file "+filePath+ "doesn't exist.");
    		}
    		fs.close();
    	}
    	
    	/*
    	 *2.2 给定HDFS中某一个目录，输出该目录下的所有文件的
    	 *读写权限、大小、创建时间、路径等信息，如果该文件是目录，
    	 *则递归输出该目录下所有文件相关信息；
    	 */
    	// 输出当前目录下的所有文件
    	public static void printFileInformation(Configuration conf, String file)throws IOException {
    		FileSystem fs = FileSystem.get(conf);
    		FileStatus[] status = fs.listStatus(new Path(file));
    		for (FileStatus s:status) {
    			System.out.println("读写权限："+s.getPermission()+"; 文件大小："+s.getBlockSize()+"; 文件路径："
    					+s.getPath()+"; 文件创建时间："+s.getModificationTime());
    		}
    		fs.close();
    	}
    	// 递归查看目录下的文件信息
    	public static void printFileInfo(Configuration conf, String filePath) throws IOException {
    		FileSystem fs = FileSystem.get(conf);
    		Path path= new Path(filePath);
    		RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(path, true);
    		while(iterator.hasNext()) {
    			FileStatus s = iterator.next();
    			System.out.println("读写权限："+s.getPermission()+"; 文件大小："+s.getBlockSize()+"; 文件路径："
    					+s.getPath()+"; 文件创建时间："+s.getModificationTime());
    		}
    		fs.close();
    	}
    	/*
    	 * 2.3 提供一个HDFS内的文件的路径，对该文件进行创建和删除操作。
    	 *      如果文件所在目录不存在，则自动创建目录；
    	 */
    	// 创建文件
    	public static void createFile(Configuration conf, String filePath) throws IOException {
    		FileSystem fs = FileSystem.get(conf);
    		Path path= new Path(filePath);
    		FSDataOutputStream ops = fs.create(path);
    		System.out.println("文件创建成功："+filePath);
    		ops.close();
    		fs.close();
    	}
    	// 2.5 删除文件
    	public static void deleteFile(Configuration conf, String filePath) throws IOException {
    		FileSystem fs = FileSystem.get(conf);
    		Path path= new Path(filePath);
    		if(fs.deleteOnExit(path)) {
    			System.out.println("文件删除成功："+filePath);
    		} else {
    			System.out.println("文件删除失败："+filePath);
    		}
    		fs.close();
    	}
    	/*
    	 * 2.4 向HDFS中指定的文件追加内容，由用户指定内容追加到原有文件的开头或结尾；
    	 */
    	// 追加内容到文件结尾
    	public static void addContentToTail(Configuration conf, String filePath, String content, boolean head) throws IOException {
    		FileSystem fs = FileSystem.get(conf);
    		Path path= new Path(filePath);
    		FSDataOutputStream ops = fs.append(path);
    		ops.write(content.getBytes());
    		if (!head) {
    			System.out.println("内容以追加到结尾。");	
    		}
    		ops.close();
    		fs.close();
    	}
    	// 移动文件到本地
    	public static void moveToLocalFile(Configuration conf, String remoteFilePath, String localFilePath) throws IOException, InterruptedException, URISyntaxException {
    		FileSystem fs = FileSystem.get(conf);			
    		Path remotePath = new Path(remoteFilePath); 
    		Path localPath = new Path(localFilePath); 
    		fs.moveToLocalFile(remotePath, localPath);
    	}
    	// 追加本地文件内容到目的文件结尾
    	public static void addFileToTail(Configuration conf, String localFilePath, String remoteFilePath) throws IOException {
    		FileSystem fs = FileSystem.get(conf);
    		Path remotePath = new Path(remoteFilePath);
    		// 创建一个本地文件(需要追加的文件)读入流
    		FileInputStream inps = new FileInputStream(localFilePath);
    		// 创建一个文件输出留，输出的内容追加到文件末尾
    		FSDataOutputStream ops = fs.append(remotePath);
    		byte[] buffer = new byte[1024];
    		int read = -1;
    		while((read = inps.read(buffer)) > 0) {
    			ops.write(buffer, 0, read);
    		}
    		ops.close();
    		inps.close();
    		fs.close();
    	}
    	// 追加内容到文件开头
    	public static void addContentToHead(Configuration conf, String filePath, String content) throws IOException, InterruptedException, URISyntaxException {
    		// 创建一个临时本地文件
    		String localTmpPath = "tmp.txt";
    		// 将要追加的文件移动到本地
    		moveToLocalFile(conf, filePath, localTmpPath);
    		// 创建一个新的HDFS文件（空的）
    		createFile(conf, filePath);
    		addContentToTail(conf, filePath, content, true);
    		addFileToTail(conf, localTmpPath, filePath);
    		System.out.println("内容以追加到开头。");
    	}
    	/*
    	 * 2.6 删除HDFS中指定的目录，由用户指定目录中如果存在文件时是否删除目录；
    	 */
    	public static void deleteDirectory(Configuration conf, String filePath) throws IOException {
    		FileSystem fs = FileSystem.get(conf);
    		Path path = new Path(filePath);
    		FileStatus[] status = fs.listStatus(path);
    		if(status.length > 0) {
    			System.out.println("该目录中存在文件，是否继续删除Y/N");
    			Scanner cin = new Scanner(System.in);
    			String flag = cin.next();
    			if(flag.equals("Y")) {
    				fs.delete(path,true);
    				System.out.println("目录已删除.");
    			}else {
    				System.out.println("用户取消删除.");
    			}
    		}else {
    			fs.delete(path,true);
    			System.out.println("目录已删除.");
    		}
    		fs.close();
    	}
    	
    	/*
    	 * 2.7 在HDFS中，将文件从源路径移动到目的路径。
    	 */
    	public static void moveFile(Configuration conf, String srcPath, String dirPath) throws IOException {
    		FileSystem fs = FileSystem.get(conf);
    		if(fs.exists(new Path(dirPath))) {
    			System.out.println("文件被占用。");
    			return;
    		}
    		if(fs.rename(new Path(srcPath), new Path(dirPath))) {
    			System.out.println("文件移动成功。");
    		} else {
    			System.out.println("文件移动失败。");
    		}
    	}
    	
    	public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
    		Scanner cin = new Scanner(System.in);
    		Configuration conf = new Configuration();
    		conf.set("fs.defaultFS", "hdfs://localhost:9000");
    		conf.set("dfs.client.block.write.replace-datanode-on-failure.enable","true");
    		conf.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
    		conf.set("dfs.replication","1");
    		// 测试文件
    		String filePath = "hadoop.txt";
    		// 创建hdfs文件
    		HdfsExpTask4.createFile(conf, filePath);
    		// 测试添加内容到文件末尾
    		HdfsExpTask4.addContentToTail(conf, filePath, "!!!hello!!!", false);
    		HdfsExpTask4.cat(conf, filePath);
    		// 测试添加内容到文件开头
    		HdfsExpTask4.addContentToHead(conf, filePath, "hello hadoop!");
    		// 文件内容打印到终端
    		HdfsExpTask4.cat(conf, filePath);
    		// 打印目标文件相关信息
    		HdfsExpTask4.printFileInformation(conf, filePath);
    		// 测试移动文件
    		HdfsExpTask4.moveFile(conf, filePath, "newHadoop.txt");
    		// 删除文件
    		HdfsExpTask4.deleteFile(conf, "newHadoop.txt");
    		// 待测试目录
    		String dirPath = "test/hadoop/aaa.txt";
    		HdfsExpTask4.createFile(conf, dirPath);
    		HdfsExpTask4.printFileInfo(conf, "test");
    		//HdfsExpTask4.deleteDirectory(conf, "test");
    		System.out.println("end.");
    	}
    }