package com.da.hbase.tool.utils;
import com.da.hbase.tool.common.Const;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
/**
* hdfs操作常用方法类
*/
public class HdfsUtils {
public static final Logger LOG= LoggerFactory.getLogger(HdfsUtils.class);
/**
* 通过ip直接连接hdfs
* @param ip
* @return
*/
public static FileSystem getFsFromIp(String ip){
FileSystem fs = null;
try {
fs=FileSystem.get(URI.create("hdfs://"+ip),new Configuration());
} catch (IOException e) {
LOG.error("此ip：{} 连接出现异常", ip);
}
return fs;
}
/**
* 检查该fs是否可用
* @param fs
* @return
*/
public static Boolean checkFs(FileSystem fs){
Boolean success=true;
if(null==fs){
return false;
}
Path path=new Path("/");
try {
RemoteIterator<FileStatus> remoteIterator= fs.listStatusIterator(path);
success=true;
} catch (IOException e) {
success=false;
}
return success;
}
/**
* 从ips中获取一个可用的fs
* @param ips
* @return
*/
public static FileSystem getAndCheckFs(String ips){
return getAndCheckFs(ips,",");
}
/**
* 从ips中获取一个可用的fs
* @param ips
* @param separator
* @return
*/
public static FileSystem getAndCheckFs(String ips,String separator){
String [] ipArr=ips.split(separator);
FileSystem fs=null;
for (String ip : ipArr) {
fs=getFsFromIp(ip);
if(checkFs(fs)){
LOG.info("此Ip：{}可连接hdfs",ip);
break;
}else{
fs=null;
}
}
if(null==fs){
LOG.error("无法连接hdfs环境，请检查网络是否可用或者ip配置是否正确，配置ips:{}",ips);
}
return fs;
}
/**
* 测试getAndCheckFs方法
*/
private static void testConnectFs(){
String ips="10.17.139.126,10.17.139.127,10.17.139.125";
FileSystem fs=getAndCheckFs(ips);
String path1="/hbase/data/default/";
Path path=new Path(path1);
try {
RemoteIterator<FileStatus> remoteIterator= fs.listStatusIterator(path);
while(remoteIterator.hasNext()){
System.out.println(remoteIterator.next().getPath());
}
} catch (IOException e) {
}
}
/**
* 查看当前路径是否存在
* @param fs
* @param path
* @return
*/
public static Boolean checkPathExist(FileSystem fs,String path){
Boolean isExist=true;
try {
isExist=fs.exists(new Path(path));
} catch (IOException e) {
isExist=false;
e.printStackTrace();
}
return isExist;
}
/**
* 递归遍历找到所有目录和文件存储在map中，文件，key：路径，value：FILE ；目录，key：路径，value：DIR
* @param fs
* @param src
*/
public static void recureScanDir(FileSystem fs,Path src, Map<Path,String> map){
try{
if(fs.isFile(src)) {
map.put(src, Const.FILE_STATUS);
}else{
map.remove(src);
RemoteIterator<FileStatus> remoteIterator= fs.listStatusIterator(src);
if(!remoteIterator.hasNext()){
map.put(src, Const.DIR_STATUS);
}else {
while (remoteIterator.hasNext()){
recureScanDir(fs,remoteIterator.next().getPath(),map);
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 目录从本地拷贝到hdfs上
* @param fs
* @param src
* @param dst
* @return
*/
public static Boolean copyFromLocal(FileSystem fs,Path src,Path dst){
Boolean success=true;
try {
if(fs.exists(dst)){
fs.delete(dst,true);
}
fs.copyFromLocalFile(false,true,src,dst);
success=true;
} catch (IOException e) {
success=false;
LOG.error("文件从本地拷贝到hdfs上，出现Io异常，导致拷贝文件失败,src:{},dst:{}",src,dst);
e.printStackTrace();
}
return success;
}
/**
*目录从hdfs上拷贝到本地
* @param fs
* @param src
* @param dst
* @return
*/
public static Boolean copyToLocal(FileSystem fs,Path src,Path dst){
Boolean success=true;
try {
if(new File(dst.toString()).exists()){
Utils.deletNotEmptyDir(new File(dst.toString()));
}
fs.copyToLocalFile(false, src, dst, true);
success=true;
} catch (IOException e) {
success=false;
LOG.error("文件从hdfs拷贝到本地，出现Io异常，导致拷贝文件失败");
e.printStackTrace();
}
return success;
}
private static void testCopyFileToLocal(){
String ips="10.17.139.126,10.17.139.127,10.17.139.125";
FileSystem fs=getAndCheckFs(ips);
String path1="/hbase/data/default/";
Path path=new Path(path1);
try {
RemoteIterator<FileStatus> remoteIterator= fs.listStatusIterator(path);
while(remoteIterator.hasNext()){
System.out.println(remoteIterator.next().getPath());
}
} catch (IOException e) {
LOG.error(e.getMessage());
}
}
/**
* 获取目录path下所有的文件名
* @param fs
* @param path
* @return
*/
public static List<String> scanDir(FileSystem fs,Path path){
List<String> list=new ArrayList<>();
try {
RemoteIterator<FileStatus> remoteIterator= fs.listStatusIterator(path);
while(remoteIterator.hasNext()){
list.add(remoteIterator.next().getPath().getName());
}
} catch (IOException e) {
LOG.error(e.getMessage());
}
return list;
}
public static void main(String[] args) {
//testConnectFs();
testCopyFileToLocal();
}
}