使用Hadoop的Java客户端API操作分布式文件系统
#获取文件系统实现
//hdfs://master01:9000/
FileSystem get(URI uri[,Configuration conf[,String user]])
//fs.defaultFS
FileSystem newInstance(URI uri[,Configuration conf[,String user]])
#从配置中获取默认URI路径
URI getDefaultUri(Configuration conf)
#打开一个指向给定HDFS文件的输入流
FSDataInputStream open(Path path[,int bufferSize])
#移动(不同的目录参数时)或重命名(相同的目录参数时)一个HDFS文件,它相当于mv命令
boolean rename(Path src,Path dest)
#创建符号链接,第一个参数是被链接的目标文件,第二个参数是符号连接,第三个参数表示符号连接是否允许创建父级目录,
//此方法调用之前必须首先调用FileSystem.enableSymlinks()以保证符号连接可用
createSymlink(Path src, Path link, boolean createParent)
#本地文件的内容直接追加到HDFS文件中
FSDataOutputStream append(Path path[,int bufferSize])
#创建一个指向给定HDFS文件的输出流,使用该输出流可以向该文件中写入内容
FSDataOutputStream create(Path path[,boolean override[,int bufferSize]])
#删除指定的HDFS文件或文件夹,第二个参数表示是否可以递归删除,方法返回是否删除成功的逻辑类型
boolean delete(Path path[,boolean recursive])
#判断给定的HDFS文件或路径是否存在
boolean exists(Path path)
#获取给定HDFS文件块的实际块尺寸(字节)
long getBlockSize(Path path)
#获取HDFS文件块的默认尺寸(字节)
long getDefaultBlockSize()
#获取文件系统的默认端口
int getDefaultPort()
#获取给定HDFS文件的尺寸
long getLength(Path path)
#获取给定HDFS文件的文件状态信息
FileStatus getFileStatus(Path path)
#获取HDFS文件的实际副本数量
int getReplication()
#设置给定文件的副本数量
boolean setReplication(Path src,short replication)
#判断给定的文件是否是文件/目录
boolean isFile(Path path)/isDirectory(Path path)
#获取HDFS文件的默认副本数量
int getDefaultReplication()
#创建指定路径的HDFS目录
boolean mkdirs(Path path[,FsPermission permission])
#获取文件系统的使用量和容量的状态
FileStatus getStatus()
#列出给定目录中的文件和文件夹
FileStatus[] listStatus(Path path[,PathFilter pf])
#第二个参数表示是否递归,如果第一个参数是文件则返回文件的状态信息,如果第一个参数是目录则表示是否递归列出该目录中的文件(不含文件夹)
RemoteIterator<LocatedFileStatus> listFiles(Path path,boolean recursive)
#获取给定HDFS文件块所在的位置
BlockLocation[] getFileBlockLocations(Path path,int start,long len)
BlockLocation[] getFileBlockLocations(FileStatus file,int start,long len)
#返回文件系统中所有文件的总尺寸
long getUsed()
#将本地单个文件拷贝到HDFS文件系统中,前两个参数表示是否需要删除源文件,是否需要覆盖目标文件
copyFromLocalFile([boolean delSrc[,boolean override]]Path src,Path dst)
#重载,将本地多个文件拷贝到HDFS文件系统中
copyFromLocalFile(boolean delSrc,boolean override,Path[] srcs,Path dst)
#将HDFS文件拷贝到本地文件系统中,第1个参数表示是否需要删除源文件
copyToLocalFile([boolean delSrc,]Path src,Path dst)
#将HDFS文件或文件夹从一个路径拷贝到另一个路径
FileUtil.copy(FileSystem srcFS,Path srcPath,FileSystem dstFS,Path dstPath,boolean delSource,boolean override,Configuration conf)
测试代码：
测试说明：
全都切换到hadoop用户下；在主机上启动hdfs集群；使用jps查看启动是否成功；
主机：master01
节点：slave01、slave02、slave03
本地机器的eclipse上的代码：（导入hadoop-2.7.3-All.jar包）
package com.mmzs.hdfs.main; import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
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.FileUtil;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator; /**
* @author hadoop
*
*/
public class HdfsMain { private static FileSystem fs;
private static Configuration conf;
static {
String uri = "hdfs://master01:9000/";
conf = new Configuration();
try {
fs = FileSystem.get(new URI(uri), conf, "hadoop");
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (URISyntaxException e) {
e.printStackTrace();
}
} public static void main(String[] args) {
// mkdir("/data");
// uploadFile("/home/hadoop/test/1.txt");//本地文件路径
// downloadFile("/data/1.txt");//HDFS上的文件路径
// openFile("/data/2.txt");
try {
// writeFile("/data/1.txt");
// rename("/data/1.txt", "/data02/1-run.txt");
// rename("/data02/1-run.txt", "/data02/1copy.txt");
createLink("/data/1.txt", "/data/info");
// deleteFile("/data/2.txt");
// listFiles("/data");
// listFiles02("/data");
// copyFile("/data/1.txt", "/data/test/");
} catch (IOException e) {
e.printStackTrace();
}
} /**
* 实现集群文件的拷贝
* @param src
* @param path
* @throws IOException
*/
public static void copyFile(String src, String path) throws IOException {
Path srcpath = new Path(src);
Path dstpath = new Path(path); Boolean flag = FileUtil.copy(fs, srcpath, fs, dstpath, false, false, conf);
if (flag) {
System.out.println("拷贝文件成功");
} else {
System.out.println("拷贝文件失败");
}
} /**
* 创建文件夹
* @param path
*/
public static void mkdir(String path) {
Path newpath = new Path(path);
try {
Boolean flag = fs.mkdirs(newpath);
if (flag) {
System.out.println("文件夹创建成功");
} else {
System.out.println("文件夹创建失败");
}
} catch (IOException e) {
e.printStackTrace();
}
} /**
* 上传本地文件到HDFS集群
* @param path
*/
public static void uploadFile(String path) {
Path localFile = new Path(path);
Path clusterPath = new Path("/data/"); try {
fs.copyFromLocalFile(false, localFile, clusterPath);
System.out.println(" 上传成功！");
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} /**
* 下载HDFS集群中的文件到本地系统中
* @param clusterPath
*/
public static void downloadFile(String clusterPath) {
Path clusterFile = new Path(clusterPath);
Path localPath = new Path("/home/hadoop/test/"); try {
fs.copyToLocalFile(false, clusterFile, localPath);
System.out.println("下载成功！");
} catch (IOException e) {
e.printStackTrace();
}
} /**
* 打开集群文件并输出文件内容
* @param clusterFile
*/
public static void openFile(String clusterFile) {
Path clusterPath = new Path(clusterFile);
FSDataInputStream fis = null; try {
fis = fs.open(clusterPath,1024);
while (fis.available()>0) {//available表示测试是否可以读
// String line = fis.readUTF();
String line = fis.readLine();
System.out.println(line);
}
} catch (IOException e) {
e.printStackTrace();
}finally {
try {
if(null != fis)
fis.close();
} catch (IOException e) {
e.printStackTrace();
}
}
} /**
* 写一个文件到HDFS集群
* @param clusterFile
* @throws IOException
*/
public static void writeFile(String clusterFile) throws IOException {
Path destFile = new Path(clusterFile);
FSDataOutputStream fos = null;
if (fs.exists(destFile) && fs.isDirectory(destFile)) {
System.out.println("参数路径指向一个目录，无法写入数据");
return;
}
//如果文件存在则以追加的方式打开，否则直接创建文件并写入内容
if (fs.exists(destFile) && fs.isFile(destFile)) {
fos = fs.append(destFile);
}else {
fos = fs.create(destFile, true);
}
try{
fos.writeUTF("he is ligang\n");
fos.writeUTF("he is 65kg\n");
fos.writeUTF("he is eat\n");
fos.writeUTF("he is running\n");
fos.flush();//一定要flush啊！！！！！！！！
System.out.println("写入完成！");
}finally{
if(null != fs) fs.close();
}
} /**
* 移动(在不同目录下)或重命名文件(同一个目录下)；
* @param path1 源文件路径
* @param path2 移动或重命名后的文件路径
* @throws IOException
*/
public static void rename(String path1, String path2) throws IOException {
Path file01 = new Path(path1);
Path file02 = new Path(path2);
if (!fs.exists(file01)) return; int index = path2.lastIndexOf('/');
String dir = path2.substring(0, index);
Path dstpath = new Path(dir); Boolean flag = null;
if (!fs.exists(dstpath)) {
flag = fs.mkdirs(dstpath);
} // if (flag) {
// return;
// } fs.rename(file01, file02);
} /**
* 创建集群文件快捷方式
* @param src
* @param link
* @throws IOException
*/
public static void createLink(String src, String link) throws IOException {
Path srcPath = new Path(src);
Path linkPath = new Path(link); //如果创建快捷方式的源文件不存在或者是一个目录就理它
if (!fs.exists(srcPath) || fs.isDirectory(srcPath)) return;
fs.createSymlink(srcPath, linkPath, true);
System.out.println("创建完成！");
} /**
* 删除集群文件或者目录
* @param path
* @throws IOException
*/
public static void deleteFile(String path) throws IOException {
Path dstPath = new Path(path);
Boolean flag = fs.delete(dstPath, true);
if (flag) {
System.out.println("删除成功");
}else {
System.out.println("删除失败");
}
} /**
* 查看给定参数集群路径下的文件或者文件夹
* @param path
* @throws IOException
*/
public static void listFiles(String path) throws IOException {
Path dstPath = new Path(path);
if(!fs.exists(dstPath) ||fs.isFile(dstPath)) {
return;
}
FileStatus[] filestatus = fs.listStatus(dstPath);
for(FileStatus fstatus:filestatus) {
if(fstatus.isFile()) {
//如果是文件打印出文件的大小尺寸（单位：byte）
long filelen = fstatus.getLen();
System.out.println(filelen);
//获取文件名
System.out.println("文件名："+fstatus.getPath().getName());
//文件的路径
System.out.println("文件路径："+fstatus.getPath().toString());
//获取文件块的信息
BlockLocation[] bls = fs.getFileBlockLocations(fstatus, 0, filelen);
//循环打印每一个块的信息
for(BlockLocation bl:bls) {
String[] hosts = bl.getHosts() ;
System.out.print("主机列表："+"\n"+"hosts:");
for (int i = 0; i < hosts.length; i++) {
System.out.print(hosts[i]+"---");
}
System.out.println();
Long blockSize = bl.getLength();
System.out.println("blockSize:"+blockSize);
String[] blockPaths = bl.getTopologyPaths();
System.out.println("块路径列表：");
for (int i = 0; i < blockPaths.length; i++) {
System.out.println(blockPaths[i]+"---");
}
}
}else {
//获取文件名
String fileName = fstatus.getPath().getName();
//获取目录名
String filePath = fstatus.getPath().toString();
System.out.println("目录名字："+fileName+"目录路径："+filePath);
}
}
} /**
* 遍历指定路径下的问夹（不含文件夹）
* @param path
* @throws IOException
*/
public static void listFiles02(String path) throws IOException {
Path dstPath = new Path(path);
if(!fs.exists(dstPath) ||fs.isFile(dstPath)) {
return;
}
RemoteIterator<LocatedFileStatus> fss = fs.listFiles(dstPath, true);
while ( fss.hasNext() ) {
LocatedFileStatus fstatus = fss.next();
if(fstatus.isFile()) {
//如果是文件打印出文件的大小尺寸（单位：byte）
long filelen = fstatus.getLen();
System.out.println("文件大小尺寸 ："+filelen);
//获取文件名
System.out.println("文件名："+fstatus.getPath().getName());
//文件的路径
System.out.println("文件路径："+fstatus.getPath().toString());
//获取文件块的信息
BlockLocation[] bls = fs.getFileBlockLocations(fstatus, 0, filelen);
//循环打印每一个块的信息
for(BlockLocation bl:bls) {
String[] hosts = bl.getHosts() ;
System.out.print("主机列表："+"\n"+"hosts:");
for (int i = 0; i < hosts.length; i++) {
System.out.print(hosts[i]+"---");
}
System.out.println();
Long blockSize = bl.getLength();
System.out.println("blockSize:"+blockSize);
String[] blockPaths = bl.getTopologyPaths();
System.out.println("块路径列表：");
for (int i = 0; i < blockPaths.length; i++) {
System.out.println(blockPaths[i]+"---");
}
}
}else {
//获取文件名
String fileName = fstatus.getPath().getName();
//获取目录名
String filePath = fstatus.getPath().toString();
System.out.println("目录名字："+fileName+"目录路径："+filePath);
}
} }
}
HdfsMain
然后在eclipse中直接诶运行即可。
然后使用：“hdfs dfs -ls /集群文件路径” 等操作来查看即可。
运行中可以查看运行状态：hdfs dfs admin -report