今天完成了大数据的一部分作业
实验目的
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
1
2
执行完上述命令不会输出结果，需要继续输入命令查看结果：
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();
}
}