hadoop hdfs java api 文件操作类

ywg

　　


package test.hadoop.util;
import java.util.Iterator;
import java.util.Map.Entry;
import org.apache.commons.lang.exception.ExceptionUtils;
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.Path;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import com.emar.adwiser.common.util.LogUtil;
import com.emar.adwiser.common.util.Logs;
/***
* HDFS 工具类
* @author zhaidw
*
*/
public class HDFSUtil {
private static Logs log = LogUtil.getLog(HDFSUtil.class);
public synchronized static FileSystem getFileSystem(String ip, int port) {
FileSystem fs = null;
String url = "hdfs://" + ip + ":" + String.valueOf(port);
Configuration config = new Configuration();
config.set("fs.default.name", url);
try {
fs = FileSystem.get(config);
} catch (Exception e) {
log.error("getFileSystem failed :"
+ ExceptionUtils.getFullStackTrace(e));
}
return fs;
}
public synchronized static void listNode(FileSystem fs) {
DistributedFileSystem dfs = (DistributedFileSystem) fs;
try {
DatanodeInfo[] infos = dfs.getDataNodeStats();
for (DatanodeInfo node : infos) {
System.out.println("HostName: " + node.getHostName() + "/n"
+ node.getDatanodeReport());
System.out.println("--------------------------------");
}
} catch (Exception e) {
log.error("list node list failed :"
+ ExceptionUtils.getFullStackTrace(e));
}
}
/**
* 打印系统配置
*
* @param fs
*/
public synchronized static void listConfig(FileSystem fs) {
Iterator<Entry<String, String>> entrys = fs.getConf().iterator();
while (entrys.hasNext()) {
Entry<String, String> item = entrys.next();
log.info(item.getKey() + ": " + item.getValue());
}
}
/**
* 创建目录和父目录
*
* @param fs
* @param dirName
*/
public synchronized static void mkdirs(FileSystem fs, String dirName) {
// Path home = fs.getHomeDirectory();
Path workDir = fs.getWorkingDirectory();
String dir = workDir + "/" + dirName;
Path src = new Path(dir);
// FsPermission p = FsPermission.getDefault();
boolean succ;
try {
succ = fs.mkdirs(src);
if (succ) {
log.info("create directory " + dir + " successed. ");
} else {
log.info("create directory " + dir + " failed. ");
}
} catch (Exception e) {
log.error("create directory " + dir + " failed :"
+ ExceptionUtils.getFullStackTrace(e));
}
}
/**
* 删除目录和子目录
*
* @param fs
* @param dirName
*/
public synchronized static void rmdirs(FileSystem fs, String dirName) {
// Path home = fs.getHomeDirectory();
Path workDir = fs.getWorkingDirectory();
String dir = workDir + "/" + dirName;
Path src = new Path(dir);
boolean succ;
try {
succ = fs.delete(src, true);
if (succ) {
log.info("remove directory " + dir + " successed. ");
} else {
log.info("remove directory " + dir + " failed. ");
}
} catch (Exception e) {
log.error("remove directory " + dir + " failed :"
+ ExceptionUtils.getFullStackTrace(e));
}
}
/**
* 上传目录或文件
*
* @param fs
* @param local
* @param remote
*/
public synchronized static void upload(FileSystem fs, String local,
String remote) {
// Path home = fs.getHomeDirectory();
Path workDir = fs.getWorkingDirectory();
Path dst = new Path(workDir + "/" + remote);
Path src = new Path(local);
try {
fs.copyFromLocalFile(false, true, src, dst);
log.info("upload " + local + " to  " + remote + " successed. ");
} catch (Exception e) {
log.error("upload " + local + " to  " + remote + " failed :"
+ ExceptionUtils.getFullStackTrace(e));
}
}
/**
* 下载目录或文件
*
* @param fs
* @param local
* @param remote
*/
public synchronized static void download(FileSystem fs, String local,
String remote) {
// Path home = fs.getHomeDirectory();
Path workDir = fs.getWorkingDirectory();
Path dst = new Path(workDir + "/" + remote);
Path src = new Path(local);
try {
fs.copyToLocalFile(false, dst, src);
log.info("download from " + remote + " to  " + local
+ " successed. ");
} catch (Exception e) {
log.error("download from " + remote + " to  " + local + " failed :"
+ ExceptionUtils.getFullStackTrace(e));
}
}
/**
* 字节数转换
*
* @param size
* @return
*/
public synchronized static String convertSize(long size) {
String result = String.valueOf(size);
if (size < 1024 * 1024) {
result = String.valueOf(size / 1024) + " KB";
} else if (size >= 1024 * 1024 && size < 1024 * 1024 * 1024) {
result = String.valueOf(size / 1024 / 1024) + " MB";
} else if (size >= 1024 * 1024 * 1024) {
result = String.valueOf(size / 1024 / 1024 / 1024) + " GB";
} else {
result = result + " B";
}
return result;
}
/**
* 遍历HDFS上的文件和目录
*
* @param fs
* @param path
*/
public synchronized static void listFile(FileSystem fs, String path) {
Path workDir = fs.getWorkingDirectory();
Path dst;
if (null == path || "".equals(path)) {
dst = new Path(workDir + "/" + path);
} else {
dst = new Path(path);
}
try {
String relativePath = "";
FileStatus[] fList = fs.listStatus(dst);
for (FileStatus f : fList) {
if (null != f) {
relativePath = new StringBuffer()
.append(f.getPath().getParent()).append("/")
.append(f.getPath().getName()).toString();
if (f.isDir()) {
listFile(fs, relativePath);
} else {
System.out.println(convertSize(f.getLen()) + "/t/t"
+ relativePath);
}
}
}
} catch (Exception e) {
log.error("list files of " + path + " failed :"
+ ExceptionUtils.getFullStackTrace(e));
} finally {
}
}
public synchronized static void write(FileSystem fs, String path,
String data) {
// Path home = fs.getHomeDirectory();
Path workDir = fs.getWorkingDirectory();
Path dst = new Path(workDir + "/" + path);
try {
FSDataOutputStream dos = fs.create(dst);
dos.writeUTF(data);
dos.close();
log.info("write content to " + path + " successed. ");
} catch (Exception e) {
log.error("write content to " + path + " failed :"
+ ExceptionUtils.getFullStackTrace(e));
}
}
public synchronized static void append(FileSystem fs, String path,
String data) {
// Path home = fs.getHomeDirectory();
Path workDir = fs.getWorkingDirectory();
Path dst = new Path(workDir + "/" + path);
try {
FSDataOutputStream dos = fs.append(dst);
dos.writeUTF(data);
dos.close();
log.info("append content to " + path + " successed. ");
} catch (Exception e) {
log.error("append content to " + path + " failed :"
+ ExceptionUtils.getFullStackTrace(e));
}
}
public synchronized static String read(FileSystem fs, String path) {
String content = null;
// Path home = fs.getHomeDirectory();
Path workDir = fs.getWorkingDirectory();
Path dst = new Path(workDir + "/" + path);
try {
// reading
FSDataInputStream dis = fs.open(dst);
content = dis.readUTF();
dis.close();
log.info("read content from " + path + " successed. ");
} catch (Exception e) {
log.error("read content from " + path + " failed :"
+ ExceptionUtils.getFullStackTrace(e));
}
return content;
}
}