您现在的位置是:首页 > 正文

实验二:熟悉常用的HDFS操作

2024-02-01 01:18:07阅读 2

实验目的

1、理解HDFS在Hadoop体系结构中的角色
2、熟悉使用HDFS操作常用的Shell命令
3、熟悉HDFS操作常用的Java API

实验平台

1、操作系统:Windows
2、Hadoop版本:3.1.3
3、JDK版本:1.8
4、Java IDE:Eclipse

实验步骤

1、编程实现以下功能,并利用Hadoop提供的Shell命令完成相同任务
1)向 HDFS 中上传任意文本文件,如果指定的文件在 HDFS 中已经存在,则由用户来指定是追加到原有文件末尾还是覆盖原有的文件;
Shell 命令
检查文件是否存在,可以使用如下命令:

cd /usr/local/hadoop
./bin/hdfs dfs -test -e text.txt

执行完上述命令不会输出结果,需要继续输入命令查看结果:

echo $?

在这里插入图片描述
如果结果显示文件已经存在,则用户可以选择追加到原来文件末尾或者覆盖原来文件,
具体命令如下:

cd /usr/local/hadoop
./bin/hdfs dfs -appendToFile local.txt text.txt #追加到原文件末尾
./bin/hdfs dfs -copyFromLocal -f local.txt text.txt #覆盖原来文件,第一种命令形式
./bin/hdfs dfs -cp -f file:///home/hadoop/local.txt text.txt#覆盖原来文件,第二种命令形式

实际上,也可以不用上述方法,而是采用如下命令来实现:

if $(hdfs dfs -test -e text.txt);
then $(hdfs dfs -appendToFile local.txt text.txt);
else $(hdfs dfs -copyFromLocal -f local.txt text.txt);
fi

在这里插入图片描述
上述代码可视为一行代码,在终端中输入第一行代码后,代码不会立即被执行,可以继续输入第 2 行代码和第 3 行代码,直到输入 fi 以后,上述代码才会真正执行。另外,上述代码中,直接使用了 hdfs 命令,而没有给出命令的路径,因为,这里假设已经配置了 PATH环境变量,把 hdfs 命令的路径“/usr/local/hadoop/bin”写入了 PATH 环境变量中。

Java 代码
我们这里需要一个jar包来实现scphttp://www.ganymed.ethz.ch/ssh2/
项目目录结构如下:
在这里插入图片描述
HDFSApi.java

package HDFSApi;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.*;

import HDFSApi.SCP;

public class HDFSApi {
	 /**
	 * 判断路径是否存在
	 */
	 public static boolean test(Configuration conf, String path) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 	return fs.exists(new Path(path));
	 }
	 /**
	 * 复制文件到指定路径
	 * 若路径已存在,则进行覆盖
	 */
	 public static void copyFromLocalFile(Configuration conf, String localFilePath, String remoteFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path localPath = new Path(localFilePath);
		 Path remotePath = new Path(remoteFilePath);
		 /* fs.copyFromLocalFile 第一个参数表示是否删除源文件,第二个参数表示是否覆盖 */
		 fs.copyFromLocalFile(false, true, localPath, remotePath);
		 fs.close();
	 }
	 /**
	 * 追加文件内容
	 */
	 public static void appendToFile(Configuration conf, String localFilePath, String remoteFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 /* 创建一个文件读入流 */
		 FileInputStream in = new FileInputStream(localFilePath);
		 /* 创建一个文件输出流,输出的内容将追加到文件末尾 */
		 FSDataOutputStream out = fs.append(remotePath);
		 /* 读写文件内容 */
		 byte[] data = new byte[1024];
		 int read = -1;
		 while ( (read = in.read(data)) > 0 ) {
			 out.write(data, 0, read);
		 }
		 out.close();
		 in.close();
		 fs.close();
	 }
	/**
	* 主函数
	 *
	*/
	public static void main(String[] args) throws IOException {
		
		SCP.get();//通过SCP将linux上的文件下载到windows上
		
		Configuration conf = new Configuration();
		conf.set("fs.default.name","hdfs://192.168.43.200:9000");
		String localFilePath = "D:\local.txt"; // 本地路径
		String remoteFilePath = "/user/hadoop/text.txt"; // HDFS 路径
		String choice = "append"; // 若文件存在则追加到文件末尾
		// String choice = "overwrite"; // 若文件存在则覆盖
		try {
			/* 判断文件是否存在 */
			Boolean fileExists = false;
			if (HDFSApi.test(conf, remoteFilePath)) {
				fileExists = true;
				System.out.println(remoteFilePath + " 已存在.");
			} else {
				System.out.println(remoteFilePath + " 不存在.");
			}
			/* 进行处理 */
			if ( !fileExists) { // 文件不存在,则上传
				HDFSApi.copyFromLocalFile(conf, localFilePath, remoteFilePath);
				System.out.println(localFilePath + " 已上传至 " + remoteFilePath);
			} else if ( choice.equals("overwrite") ) { // 选择覆盖
				HDFSApi.copyFromLocalFile(conf, localFilePath, remoteFilePath);
				System.out.println(localFilePath + " 已覆盖 " + remoteFilePath);
			} else if ( choice.equals("append") ) { // 选择追加
				HDFSApi.appendToFile(conf, localFilePath, remoteFilePath);
				System.out.println(localFilePath + " 已追加至 " + remoteFilePath);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

SCP.java

package HDFSApi;
import java.io.IOException;
import ch.ethz.ssh2.Connection;
import ch.ethz.ssh2.SCPClient;
public class SCP {
	public static void get() throws IOException{
		Connection conn = new Connection("192.168.43.200");
		conn.connect();
		boolean isAuthenticated = conn.authenticateWithPassword("root", "123456");
		if (isAuthenticated == false)
			throw new IOException("Authentication failed.");
		SCPClient client = new SCPClient(conn);
		client.get("/root/local.txt", "D:\");
		conn.close();
	}
}

log4j.properties

log4j.rootLogger=info,consolePrint,errorFile,logFile

log4j.appender.consolePrint.Encoding = UTF-8
log4j.appender.consolePrint = org.apache.log4j.ConsoleAppender
log4j.appender.consolePrint.Target = System.out
log4j.appender.consolePrint.layout = org.apache.log4j.PatternLayout
log4j.appender.consolePrint.layout.ConversionPattern=%d %p [%c] - %m%n

log4j.appender.logFile.Encoding = UTF-8
log4j.appender.logFile = org.apache.log4j.DailyRollingFileAppender
log4j.appender.logFile.File = D:/eclipse/log4j_properties_runlog/WorldCount_demo_run.log
log4j.appender.logFile.Append = true
log4j.appender.logFile.Threshold = info
log4j.appender.logFile.layout = org.apache.log4j.PatternLayout
log4j.appender.logFile.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss}  [ %t:%r ] - [ %p ]  %m%n

log4j.appender.errorFile.Encoding = UTF-8
log4j.appender.errorFile = org.apache.log4j.DailyRollingFileAppender
log4j.appender.errorFile.File = D:/eclipse/log4j_properties_runlog/WorldCount_demo_error.log
log4j.appender.errorFile.Append = true
log4j.appender.errorFile.Threshold = ERROR
log4j.appender.errorFile.layout = org.apache.log4j.PatternLayout
log4j.appender.errorFile.layout.ConversionPattern =%-d{yyyy-MM-dd HH:mm:ss}  [ %t:%r ] - [ %p ]  %m%n

在这里插入图片描述
2)从 HDFS 中下载指定文件,如果本地文件与要下载的文件名称相同,则自动对下载的文件重命名
shell命令

if $(hdfs dfs -test -e file:///home/hadoop/text.txt);
then $(hdfs dfs -copyToLocal text.txt ./text2.txt);
else $(hdfs dfs -copyToLocal text.txt ./text.txt);
fi

Java代码
在这里插入图片描述
HDFSApi

package HDFSApi;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.*;

public class HDFSApi {
	 /**
	 * 下载文件到本地
	 * 判断本地路径是否已存在,若已存在,则自动进行重命名
	 */
	public static void copyToLocal(Configuration conf, String remoteFilePath, String localFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 File f = new File(localFilePath);
		 /* 如果文件名存在,自动重命名(在文件名后面加上 _0, _1 ...) */
		 if (f.exists()) {
			 System.out.println(localFilePath + " 已存在.");
			 Integer i = 0;
			 while (true) {
				 f = new File(localFilePath + "_" + i.toString());
				 if (!f.exists()) {
					 localFilePath = localFilePath + "_" + i.toString();
					 break;
				 }
				 i++;
			 }
			 System.out.println("将重新命名为: " + localFilePath);
		 }
		 
		 // 下载文件到本地
		 Path localPath = new Path(localFilePath);
		 fs.copyToLocalFile(remotePath, localPath);
		 fs.close();
	 }
	 
	/**
	* 主函数
	*/
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		conf.set("fs.default.name","hdfs://192.168.43.200:9000");
		String localFilePath = "D://text.txt"; // 本地路径
		String remoteFilePath = "/user/hadoop/text.txt"; // HDFS 路径
		try {
			HDFSApi.copyToLocal(conf, remoteFilePath, localFilePath);
			System.out.println("下载完成");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

在这里插入图片描述

(3) 将 HDFS 中指定文件的内容输出到终端中
shell命令

hdfs dfs -cat text.txt

在这里插入图片描述

java命令
在这里插入图片描述

package HDFSApi;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.*;
	public class HDFSApi {
	 /**
	 * 读取文件内容
	 */
	 public static void cat(Configuration conf, String remoteFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 FSDataInputStream in = fs.open(remotePath);
		 BufferedReader d = new BufferedReader(new InputStreamReader(in));
		 String line = null;
		 while ( (line = d.readLine()) != null ) {
			 System.out.println(line);
		 }
		 d.close();
		 in.close();
		 fs.close();
	 }
	 
	/**
	* 主函数
	*/
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		conf.set("fs.default.name","hdfs://192.168.43.200:9000");
		String remoteFilePath = "/user/hadoop/text.txt"; // HDFS 路径
		try {
			System.out.println("读取文件: " + remoteFilePath);
			HDFSApi.cat(conf, remoteFilePath);
			System.out.println("
 读取完成");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

在这里插入图片描述
(4) 显示 HDFS 中指定的文件的读写权限、大小、创建时间、路径等信息;
shell命令

 hdfs dfs -ls -h text.txt

在这里插入图片描述
Java代码
在这里插入图片描述

package HDFSApi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.*;
import java.text.SimpleDateFormat;
public class HDFSApi {
	 /**
	 * 显示指定文件的信息
	 */
	 public static void ls(Configuration conf, String remoteFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 FileStatus[] fileStatuses = fs.listStatus(remotePath);
		 for (FileStatus s : fileStatuses) {
			 System.out.println("路径: " + s.getPath().toString());
			 System.out.println("权限: " + s.getPermission().toString());
			 System.out.println("大小: " + s.getLen());
			 /* 返回的是时间戳,转化为时间日期格式 */
			 Long timeStamp = s.getModificationTime();
			 SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			 String date = format.format(timeStamp); 
			 System.out.println("时间: " + date);
		 }
		 fs.close();
	 }
	 
	/**
	* 主函数
	*/
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		 conf.set("fs.default.name","hdfs://192.168.43.200:9000");
		String remoteFilePath = "/user/hadoop/text.txt"; // HDFS 路径
		try {
			System.out.println("读取文件信息: " + remoteFilePath);
			HDFSApi.ls(conf, remoteFilePath);
			System.out.println("
 读取完成");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

在这里插入图片描述
(5) 给定 HDFS 中某一个目录,输出该目录下的所有文件的读写权限、大小、创建时
间、路径等信息,如果该文件是目录,则递归输出该目录下所有文件相关信息;
shell命令

cd /usr/local/hadoop
./bin/hdfs dfs -ls -R -h /user/hadoop

在这里插入图片描述
java代码
在这里插入图片描述

package HDFSApi;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.*;
import java.text.SimpleDateFormat;
public class HDFSApi {
	 /**
	 * 显示指定文件夹下所有文件的信息(递归)
	 */
	public static void lsDir(Configuration conf, String remoteDir) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path dirPath = new Path(remoteDir);
		 /* 递归获取目录下的所有文件 */
		 RemoteIterator<LocatedFileStatus> remoteIterator = fs.listFiles(dirPath, true);
		 /* 输出每个文件的信息 */
		while (remoteIterator.hasNext()) {
			 FileStatus s = remoteIterator.next();
			 System.out.println("路径: " + s.getPath().toString());
			 System.out.println("权限: " + s.getPermission().toString());
			 System.out.println("大小: " + s.getLen());
			 /* 返回的是时间戳,转化为时间日期格式 */
			 Long timeStamp = s.getModificationTime();
			 SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			 String date = format.format(timeStamp); 
			 System.out.println("时间: " + date);
			 System.out.println();
		}
		 fs.close();
	} 
	 
	/**
	* 主函数
	*/
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		 conf.set("fs.default.name","hdfs://192.168.43.200:9000");
		String remoteDir = "/user/hadoop"; // HDFS 路径
		try {
			System.out.println("(递归)读取目录下所有文件的信息: " + remoteDir);
			HDFSApi.lsDir(conf, remoteDir);
			System.out.println("读取完成");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

在这里插入图片描述
(6) 提供一个 HDFS 内的文件的路径,对该文件进行创建和删除操作。如果文件所在
目录不存在,则自动创建目录;
shell命令

if $(hdfs dfs -test -d dir1/dir2);
then $(hdfs dfs -touchz dir1/dir2/filename);
else $(hdfs dfs -mkdir -p dir1/dir2 && hdfs dfs -touchz dir1/dir2/filename);
fi
hdfs dfs -rm dir1/dir2/filename #删除文件

在这里插入图片描述
java代码
在这里插入图片描述

package HDFSApi;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.*;
public class HDFSApi {
	 /**
	 * 判断路径是否存在
	 */
	public static boolean test(Configuration conf, String path) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 return fs.exists(new Path(path));
	}
	 /**
	 * 创建目录
	 */
	public static boolean mkdir(Configuration conf, String remoteDir) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path dirPath = new Path(remoteDir);
		 boolean result = fs.mkdirs(dirPath);
		 fs.close();
		 return result;
	}
	 /**
	 * 创建文件
	 */
	public static void touchz(Configuration conf, String remoteFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 FSDataOutputStream outputStream = fs.create(remotePath);
		 outputStream.close();
		 fs.close();
	}
	 
	 /**
	 * 删除文件
	 */
	public static boolean rm(Configuration conf, String remoteFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 boolean result = fs.delete(remotePath, false);
		 fs.close();
		 return result;
	}
	/**
	* 主函数
	*/
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		 conf.set("fs.default.name","hdfs://192.168.43.200:9000");
		String remoteFilePath = "/user/hadoop/input/text.txt"; // HDFS 路径
		String remoteDir = "/user/hadoop/input"; // HDFS 路径对应的目录
		try {
			/* 判断路径是否存在,存在则删除,否则进行创建 */
			if ( HDFSApi.test(conf, remoteFilePath) ) {
				HDFSApi.rm(conf, remoteFilePath); // 删除
				System.out.println("删除路径: " + remoteFilePath);
			} else {
				if ( !HDFSApi.test(conf, remoteDir) ) { // 若目录不存在,则进行创建
					HDFSApi.mkdir(conf, remoteDir);
					System.out.println("创建文件夹: " + remoteDir);
				}
				HDFSApi.touchz(conf, remoteFilePath);
				System.out.println("创建路径: " + remoteFilePath);
			}
		} catch (Exception e) {
				e.printStackTrace();
		}
	}
}

在这里插入图片描述
(7) 提供一个 HDFS 的目录的路径,对该目录进行创建和删除操作。创建目录时,如
果目录文件所在目录不存在,则自动创建相应目录;删除目录时,由用户指定
当该目录不为空时是否还删除该目录;
shell命令
创建目录的命令如下:

 hdfs dfs -mkdir -p dir1/dir2

删除目录的命令如下:

hdfs dfs -rmdir dir1/dir2

上述命令执行以后,如果目录非空,则会提示 not empty,删除操作不会执行。如果要
强制删除目录,可以使用如下命令:

hdfs dfs -rm -R dir1/dir2

在这里插入图片描述
java代码
在这里插入图片描述

package HDFSApi;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.*;
public class HDFSApi {
	 /**
	 * 判断路径是否存在
	 */
	public static boolean test(Configuration conf, String path) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 return fs.exists(new Path(path));
	}
	 /**
	 * 判断目录是否为空
	 * true: 空,false: 非空
	 */
	public static boolean isDirEmpty(Configuration conf, String remoteDir) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path dirPath = new Path(remoteDir);
		 RemoteIterator<LocatedFileStatus> remoteIterator = fs.listFiles(dirPath, true);
		 return !remoteIterator.hasNext();
	}
	 /**
	 * 创建目录
	 */
	public static boolean mkdir(Configuration conf, String remoteDir) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path dirPath = new Path(remoteDir);
		 boolean result = fs.mkdirs(dirPath);
		 fs.close();
		 return result;
	}
	 
	 /**
	 * 删除目录
	 */
	public static boolean rmDir(Configuration conf, String remoteDir) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path dirPath = new Path(remoteDir);
		 /* 第二个参数表示是否递归删除所有文件 */
		 boolean result = fs.delete(dirPath, true);
		 fs.close();
		 return result;
	}
	/**
	* 主函数
	*/
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		 conf.set("fs.default.name","hdfs://192.168.43.200:9000");
		String remoteDir = "/user/hadoop/input"; // HDFS 目录
		Boolean forceDelete = false; // 是否强制删除
		try {
			/* 判断目录是否存在,不存在则创建,存在则删除 */
			if ( !HDFSApi.test(conf, remoteDir) ) {
				HDFSApi.mkdir(conf, remoteDir); // 创建目录
				System.out.println("创建目录: " + remoteDir);
			} else {
				if ( HDFSApi.isDirEmpty(conf, remoteDir) || forceDelete ) { // 目录为空或强制删除
					HDFSApi.rmDir(conf, remoteDir);
					System.out.println("删除目录: " + remoteDir);
				} else { // 目录不为空
					System.out.println("目录不为空,不删除: " + remoteDir);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

在这里插入图片描述
(8) 向HDFS中指定的文件追加内容,由用户指定内容追加到原有文件的开头或结尾;
shell命令
追加到原文件末尾的命令如下:

 hdfs dfs -appendToFile local.txt text.txt

在这里插入图片描述

追加到原文件的开头,在 HDFS 中不存在与这种操作对应的命令,因此,无法使用一条
命令来完成。可以先移动到本地进行操作,再进行上传覆盖,具体命令如下:

hdfs dfs -get text.txt
cat text.txt >> local.txt
hdfs dfs -copyFromLocal -f local.txt text.txt

在这里插入图片描述
java代码
在这里插入图片描述

package HDFSApi;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.*;
public class HDFSApi {
	 /**
	 * 判断路径是否存在
	 */
	public static boolean test(Configuration conf, String path) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 return fs.exists(new Path(path));
	}
	 /**
	 * 追加文本内容
	 */
	 public static void appendContentToFile(Configuration conf, String content, String remoteFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 /* 创建一个文件输出流,输出的内容将追加到文件末尾 */
		 FSDataOutputStream out = fs.append(remotePath);
		 out.write(content.getBytes());
		 out.close();
		 fs.close();
	}
	 /**
	 * 追加文件内容
	 */
	 public static void appendToFile(Configuration conf, String localFilePath, String remoteFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 /* 创建一个文件读入流 */
		 FileInputStream in = new FileInputStream(localFilePath);
		 /* 创建一个文件输出流,输出的内容将追加到文件末尾 */
		 FSDataOutputStream out = fs.append(remotePath);
		 /* 读写文件内容 */
		 byte[] data = new byte[1024];
		 int read = -1;
		 if(in!=null){
			while ( (read = in.read(data)) > 0 ) {
					out.write(data, 0, read);
			}
		 }
		 out.close();
		 in.close();
		 fs.close();
	}
	 /**
	 * 移动文件到本地
	 * 移动后,删除源文件
	 */
	 public static void moveToLocalFile(Configuration conf, String remoteFilePath, String localFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 Path localPath = new Path(localFilePath);
		 fs.moveToLocalFile(remotePath, localPath);
	}
	 
	 /**
	 * 创建文件
	 */
	 public static void touchz(Configuration conf, String remoteFilePath) throws IOException { 
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 FSDataOutputStream outputStream = fs.create(remotePath);
		 outputStream.close();
		 fs.close();
	 }
 
	/**
	* 主函数
	*/
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		 conf.set("fs.default.name","hdfs://192.168.43.200:9000");
		String remoteFilePath = "/user/hadoop/text.txt"; // HDFS 文件
		String content = "新追加的内容
";
		//String choice = "after"; //追加到文件末尾
		 String choice = "before"; // 追加到文件开头
		try {
			/* 判断文件是否存在 */
			if ( !HDFSApi.test(conf, remoteFilePath) ) {
				System.out.println("文件不存在: " + remoteFilePath);
			} else {
				if ( choice.equals("after") ) { // 追加在文件末尾
					HDFSApi.appendContentToFile(conf, content, remoteFilePath);
					System.out.println("已追加内容到文件末尾" + remoteFilePath);
				} else if ( choice.equals("before") ) { // 追加到文件开头
					/* 没有相应的 api 可以直接操作,因此先把文件移动到本地*/
					/*创建一个新的 HDFS,再按顺序追加内容 */
					String localTmpPath = "/user/hadoop/tmp.txt";
					// 移动到本地
					HDFSApi.moveToLocalFile(conf, remoteFilePath, localTmpPath);
					 // 创建一个新文件
					HDFSApi.touchz(conf, remoteFilePath); 
					 // 先写入新内容
					HDFSApi.appendContentToFile(conf, content, remoteFilePath);
					 // 再写入原来内容
					HDFSApi.appendToFile(conf, localTmpPath, remoteFilePath); 
					System.out.println("已追加内容到文件开头: " + remoteFilePath);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

在这里插入图片描述
注:这里选择追加到末尾没有问题,如果追加到头部会产生一个问题

Failed to replace a bad datanode on the existing pipeline due to no more good datanodes being available to try.

在这里插入图片描述

FSDataOutputStream out = fs.append(remotePath);
1):如果集群中的文件是空文件,则能追加进去。
2):如果集群中的文件有内容,则追加的时候会报错。
原因是因为hadoop环境中只有两个datanode slave1,slave2,太少的datanode节点导致追加操作错误太多。
解决方法
1、我们将datanode扩展至三个或以上,我们可以将master也加入datanode,这样就有了三个节点。

stop-all.sh  #停下hadoop
vi /usr/local/hadoop/etc/hadoop/workers  #增加master节点

在这里插入图片描述

start-all.sh  #启动hadoop

2、在管道中没有可用datanode时,生成新的datanode
我们需要在java代码中添加conf

/**
 * dfs.client.block.write.replace-datanode-on-failure.enable=true
 * 如果在写入管道中存在一个DataNode或者网络故障时,
 * 那么DFSClient将尝试从管道中删除失败的DataNode,
 * 然后继续尝试剩下的DataNodes进行写入。
 * 结果,管道中的DataNodes的数量在减少。
 * enable :启用特性,disable:禁用特性
 * 该特性是在管道中添加新的DataNodes。
 * 当集群规模非常小时,例如3个节点或更少时,集群管理员可能希望将策略在默认配置文件里面设置为NEVER或者禁用该特性。
 * 否则,因为找不到新的DataNode来替换,用户可能会经历异常高的管道写入错误,导致追加文件操作失败
 */
conf.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
 
/**
 * dfs.client.block.write.replace-datanode-on-failure.policy=DEFAULT
 * 这个属性只有在dfs.client.block.write.replace-datanode-on-failure.enable设置true时有效:
 * ALWAYS :当一个存在的DataNode被删除时,总是添加一个新的DataNode
 * NEVER  :永远不添加新的DataNode
 * DEFAULT:副本数是r,DataNode的数时n,只要r >= 3时,或者floor(r/2)大于等于n时
 * r>n时再添加一个新的DataNode,并且这个块是hflushed/appended
 */
conf.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");

(9) 删除 HDFS 中指定的文件
shell命令

hdfs dfs -rm text.txt

java代码
在这里插入图片描述

package HDFSApi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.*;
public class HDFSApi {
	 /**
	 * 删除文件
	 */
	public static boolean rm(Configuration conf, String remoteFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 boolean result = fs.delete(remotePath, false);
		 fs.close();
		 return result;
	}
	 
	/**
	* 主函数
	*/
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		 conf.set("fs.default.name","hdfs://192.168.43.200:9000");
		String remoteFilePath = "/user/hadoop/text.txt"; // HDFS 文件
		try {
			if ( HDFSApi.rm(conf, remoteFilePath) ) {
				System.out.println("文件删除: " + remoteFilePath);
			} else {
				System.out.println("操作失败(文件不存在或删除失败)");
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

在这里插入图片描述

(10) 在 HDFS 中,将文件从源路径移动到目的路径
shell命令

hdfs dfs -mv text.txt text2.txt

java代码

package HDFSApi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.*;
public class HDFSApi {
	 /**
	 * 移动文件
	 */
	 public static boolean mv(Configuration conf, String remoteFilePath, String remoteToFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path srcPath = new Path(remoteFilePath);
		 Path dstPath = new Path(remoteToFilePath);
		 boolean result = fs.rename(srcPath, dstPath);
		 fs.close();
		 return result;
	}
	 
	/**
	* 主函数
	*/
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		 conf.set("fs.default.name","hdfs://192.168.43.200:9000");
		String remoteFilePath = "hdfs:///user/hadoop/text.txt"; // 源文件 HDFS 路径
		String remoteToFilePath = "hdfs:///user/hadoop/new.txt"; // 目的 HDFS 路径
		try {
			if ( HDFSApi.mv(conf, remoteFilePath, remoteToFilePath) ) {
				System.out.println(" 将文件 " + remoteFilePath + " 移动到 " + remoteToFilePath);
			} else {
				System.out.println("操作失败(源文件不存在或移动失败)");
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

在这里插入图片描述

(二)编程实现一个类“MyFSDataInputStream”

该类继承“org.apache.hadoop.fs.FSDataInputStream”,要求如下:实现按行读取 HDFS 中指定文件的方法“readLine()”,如果读到文件末尾,则返回空,否则返回文件一行的文本。

package HDFSApi;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.*;
public class MyFSDataInputStream extends FSDataInputStream {
	public MyFSDataInputStream(InputStream in) {
		super(in);
	}
	/**
	 * 实现按行读取
	 * 每次读入一个字符,遇到"
"结束,返回一行内容
	 */
	public static String readline(BufferedReader br) throws IOException {
		char[] data = new char[1024];
		int read = -1;
		int off = 0; 
		// 循环执行时,br 每次会从上一次读取结束的位置继续读取
		//因此该函数里,off 每次都从 0 开始
		while ( (read = br.read(data, off, 1)) != -1 ) {
			if (String.valueOf(data[off]).equals("
") ) {
				off += 1;
				break;
			}
			off += 1;
		}
		if (off > 0) {
			return String.valueOf(data);
		} else {
			return null;
		}
	}
	/**
	 * 读取文件内容
	 */
	public static void cat(Configuration conf, String remoteFilePath) throws IOException {
		 FileSystem fs = FileSystem.get(conf);
		 Path remotePath = new Path(remoteFilePath);
		 FSDataInputStream in = fs.open(remotePath);
		 BufferedReader br = new BufferedReader(new InputStreamReader(in));
		 String line = null;
		 while ( (line = MyFSDataInputStream.readline(br)) != null ) {
			 System.out.println(line);
		 }
		 br.close();
		 in.close();
		 fs.close();
	}
	/**
	* 主函数
	*/
	public static void main(String[] args) {
		Configuration conf = new Configuration();
		 conf.set("fs.default.name","hdfs://192.168.43.200:9000");
		String remoteFilePath = "/user/hadoop/text.txt"; // HDFS 路径
		try {
			MyFSDataInputStream.cat(conf, remoteFilePath);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

在这里插入图片描述

(三)输出 HDFS 中指定文件的文本到终端中

查看 Java 帮助手册或其它资料,用“java.net.URL”和“org.apache.hadoop.fs.FsURLSt
reamHandlerFactory”编程完成输出 HDFS 中指定文件的文本到终端中。

package HDFSApi;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import java.io.*;
import java.net.URL;
public class HDFSApi {
	static{ 
		URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()); 
	 }
	 /**
	* 主函数
	*/
	public static void main(String[] args) throws Exception {
		String remoteFilePath = "hdfs://192.168.43.200:9000/user/hadoop/text.txt"; // HDFS 文件
		InputStream in = null; 
		 try{ 
			 /* 通过 URL 对象打开数据流,从中读取数据 */
			 in = new URL(remoteFilePath).openStream(); 
		 	IOUtils.copyBytes(in,System.out,4096,false); 
		 } finally{ 
			 IOUtils.closeStream(in); 
		 } 
	}
}

在这里插入图片描述

网站文章

  • 【高并发】多线程环境下如何保证共享变量的安全性?冰河建议你使用ThreadLocal,一文带你彻底搞懂ThreadLocal

    【高并发】多线程环境下如何保证共享变量的安全性?冰河建议你使用ThreadLocal,一文带你彻底搞懂ThreadLocal

    ThreadLocal是JDK提供的,支持线程本地变量。也就是说,如果我们创建了一个ThreadLocal变量,则访问这个变量的每个线程都会有这个变量的一个本地副本。如果多个线程同时对这个变量进行读写操作时,实际上操作的是线程自己本地内存中的变量,从而避免了线程安全的问题。

    2024-02-01 01:18:01
  • 2021-10-13-草稿纸

    2021-10-13-草稿纸

    https://www.cnblogs.com/nc923/p/11288072.htmllayui-官方demo: <!DOCTYPE html><html lang="en"><head> <...

    2024-02-01 01:17:53
  • “接口”的定义及其与“抽象类”的区别

    我们知道一个有抽象方法的类是抽象类,而当一个类中全是抽象方法时,就可以定义为接口(interface) 接口命名通常以“I”开头; 接口中的方法默认有public abstract(所以可以省略); 接口中的变量默认有public static final(静态常量),因此可以通过接口名直接调用,例如: public interface IMammal{ public abstract vo...

    2024-02-01 01:17:26
  • websocket 的创建与关闭

    websocket 的创建与关闭。

    2024-02-01 01:17:18
  • C#读Visio模型数据

    C#读Visio模型数据

    第一步:安装Visio安装相应版本的Visio(以64位为例),在安装过程中经常遇到下面的问题解决方法:按“win+R”快捷键,打开“运行”,输入“regedit”,打开注册表,找到【HKEY_CLASSES_ROOT】àInstalleràProducts备份注册表,然后将上面Office 16 Click-to-run Extensiblity Component删除,然后...

    2024-02-01 01:17:11
  • [学习记录] Redis 5. 事务和锁机制,秒杀案例 Demo

    [学习记录] Redis 5. 事务和锁机制,秒杀案例 Demo

    [学习记录] Redis 5. 事务和锁机制,秒杀案例 Demo

    2024-02-01 01:17:07
  • P5735 【深基7.例1】距离函数

    P5735 【深基7.例1】距离函数给出平面坐标上不在一条直线上三个点坐标 (x1,y1),(x2,y2),(x3,y3)(x_1,y_1),(x_2,y_2),(x_3,y_3)(x1,y1),(x...

    2024-02-01 01:16:37
  • C# - JSON详解

    C# - JSON详解

    最近在做微信开发时用到了一些json的问题,就是把微信返回回来的一些json数据做一些处理,但是之前json掌握的不好,浪费了好多时间在查找一些json有关的转换问题,我所知道的方法只有把json序列化和反序列化一下,但是太麻烦了我觉得,所以就在找一些更简单又方便使用的方法。也许这个会有用吧,所以先放到这以后能用到的。原文出处:http://www.cnblogs.com/mcgra...

    2024-02-01 01:16:29
  • yum安装mongodb报错

    yum安装mongodb报错

    今天在尝试yum安装mongodb时,发现了问题:这是因为你以前用的是CENTOS现在是redhat 红帽的yum安装软件的时候要验证的看是不是红帽的软件,是红帽的软件可以安装不是就失败.因此,我们需要将gpgcheck=1改成gpgcheck=0即可。gpgcheck=1表示需要验证,0表示不需要验证。成功!...

    2024-02-01 01:16:13
  • 用nginx反向代理Jenkins遇到的testForReverseProxySetup问题

    用nginx反向代理Jenkins遇到的testForReverseProxySetup问题

    又一次开始了Jenkins征程,其实以前我就遇到了这个问题,如图 你说你报这个错误鬼知道是为什么,当然了,我们也不能太苛求,Jenkins怎么可能知道具体是什么问题呢?算啦,我们自己去看Jenkins的日志吧,我发现 WARNING: http://jenkins.tangxuyang.cn/manage vs. http: 然后又结合chrome的F12,如下 就是这个请求没有正...

    2024-02-01 01:15:44