package com.db.hadoop.hdfs;



import java.io.ByteArrayInputStream;

import java.io.IOException;

import java.io.OutputStream;

import java.net.URI;

import java.text.SimpleDateFormat;

import java.util.Date;



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.Path;

import org.apache.hadoop.io.IOUtils;

import org.apache.hadoop.util.Progressable;

import org.apache.log4j.Logger;



public class HDFSUtil {
private static final Logger LOG = Logger.getLogger(HDFSUtil.class);  //日志记录器对象
private static FileSystem fileSystem;   //文件系统对象

static{ 
Configuration conf = new Configuration();  //hadoop的配置文件对象
URI uri = URI.create("hdfs://master:9000/");  //hadoop分布式文件系统master的uri
try {
fileSystem = FileSystem.get(uri, conf, "hadoop"); //根据配置个远程地址创建文件系统
LOG.debug("创建文件系统对象成功。。。。。。。。");
} catch (IOException | InterruptedException e) {
e.printStackTrace();
LOG.error("创建文件系统对象失败。。。。。。。。。");
} 
}


/*查询操作
*打开指定上的文件，导出内容到指定位置
*@param inHdfsPath：Hdfs上的文件
*@param out：导出的位置，当为null时在控制台输出
*/
public static void readFile(String inHdfsPath, OutputStream out){
FSDataInputStream in = null;
try {
in = fileSystem.open(new Path(inHdfsPath));
LOG.debug("打开文件系统上的文件成功。。。。。");
} catch (IOException | IllegalArgumentException e) {
LOG.error("打开文件系统上的文件失败。。。。。",e);
throw new RuntimeException("打开文件系统上的文件失败.......");
}

try {
IOUtils.copyBytes(in, out, 4096, true);
LOG.debug("拷贝文件数据流成功。。。。。。。");
}catch (IOException e) {
LOG.error("拷贝文件数据流失败。。。。。。。");
}
}


/*取到文件系统上的文件数据信息
*@param inHdfsPath:HDFS上的文件
*@return 返回元数据信息对象
*/
public static FileStatus getFileStatus(String isHdfsPath){
try {
return fileSystem.getFileStatus(new Path(isHdfsPath));
} catch (IOException | IllegalArgumentException e) {
LOG.error("取到文件系统上面的文件信息失败",e);
throw new RuntimeException("取到文件系统上面的文件信息失败......");
} 
}


/*取到文件系统上的文件元数据信息的具体详情
*@param inHdfsPath:Hdfs上的文件
*@return 返回元数据信息的具体详情对象
*/
public static BlockLocation[] getBlockLocation(String inHdfsPath){
FileStatus fileStatus = getFileStatus(inHdfsPath);
try {
return fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
} catch (IOException e) {
LOG.error("取到文件系统上面的文件详情信息失败",e);
throw new RuntimeException("取到文件系统上面的文件详情信息失败.......");
}

}


/*创建文件
*@param dirPath:要创建的文件的目录
*@return 是否创建成功
*/
public static boolean createDir(String dirPath){
try {
if(fileSystem.exists(new Path(dirPath))){
LOG.debug("在文件系统中存在该目录");
return true;
}else{
//默认目录权限755 ，默认文件权限是644
return fileSystem.mkdirs(new Path(dirPath));
}
} catch (IOException | IllegalArgumentException e) {
LOG.error("在文件系统中创建目录失败",e);
throw new RuntimeException("在文件系统中创建目录失败.....");
} 
}


/*删除文件
*@param filePath:要删除的文件
*@param recursive:要删除的目录文件
*@return 是否成功
*/
public static boolean del(String filePath, boolean recursive){
try {
return fileSystem.delete(new Path(filePath), recursive);
} catch (IOException | IllegalArgumentException e) {
LOG.error("删除文件系统上面的文件失败",e);
throw new RuntimeException("删除文件系统上面的文件失败......");
} 
}


/*保存文件
*@param saveFilePath:保存文件的路径
*@param content:要保存的内容
*@param isAppend:是否追加 
*return 是否成功
*/
public static boolean save(String saveFilePath, String content, boolean isAppend){
FSDataOutputStream out = null;
try {
if(fileSystem.exists(new Path(saveFilePath))){
//文件存在，追加内容
out = fileSystem.append(new Path(saveFilePath), 4096, new Progressable(){
@Override
public void progress() {
System.out.println(">>");

}
});
}else{
//文件不存在
out = fileSystem.create(new Path(saveFilePath), true, 4096, new Progressable(){
@Override
public void progress() {
System.out.println("....");

}

});
}
ByteArrayInputStream in = new ByteArrayInputStream(content.getBytes());
IOUtils.copyBytes(in, out, 4096, true);
System.out.println("写入数据完成........");
return true;
} catch (IOException | IllegalArgumentException e) {
LOG.error("保存到数据文件系统上失败",e);
throw new RuntimeException("保存到数据文件系统上失败.....");
}

}


/*从文件系统下载文件
*@param srcPath:要下载的文件
*@param dstPath:要下载到的文件系统路径
*@return 是否成功
*/
public static boolean download(String srcPath, String dstPath){
try {
fileSystem.copyToLocalFile(new Path(srcPath), new Path(dstPath));
return true;
} catch (IOException | IllegalArgumentException e) {
LOG.error("从文件系统上下载文件失败",e);
throw new RuntimeException("从文件系统上下载文件失败.....");
}
}


/*上传文件到文件系统
*@param srcPath:要上传的文件
*@param dstPath:要上传到的文件系统路径
*@param overwrite:是否覆盖文件
*@return 是否成功
*/
public static boolean upload(String srcPath,String dstPath,boolean overwrite){
try {
fileSystem.copyFromLocalFile(new Path(srcPath), new Path(dstPath));
return true;
} catch (IOException | IllegalArgumentException e) {
LOG.error("上传文件到文件系统上失败",e);
throw new RuntimeException("上传文件到文件系统上失败.....");
}

}


/*字节数据的单位转换
*@param num 字节数大小
*@return 带单位的字节数大小
*/
public static String showSize(double num){
//  B 1024KB  1024MB  1024GB  1024PB  1024ZB  1024EB
if(num < Math.pow(1024, 1)){
return num + "B";
}else if(num > Math.pow(1024, 1) && num < Math.pow(1024, 2)){
return num / Math.pow(1024, 1) + "KB";
}else if(num > Math.pow(1024, 2) && num < Math.pow(1024, 3)){
return num / Math.pow(1024, 2) + "MB";
}else if(num > Math.pow(1024, 3) && num < Math.pow(1024, 4)){
return num / Math.pow(1024, 3) + "GB";
}else if(num > Math.pow(1024, 4) && num < Math.pow(1024, 5)){
return num / Math.pow(1024, 4) + "PB";
}else if(num > Math.pow(1024, 5) && num < Math.pow(1024, 6)){
return num / Math.pow(1024, 5) + "ZB";
}else if(num > Math.pow(1024, 6) && num < Math.pow(1024, 7)){
return num / Math.pow(1024, 6) + "EB";
}else{
return "很大很大很大";
}
}


/*打印文件系统上面的文件信息
*@param isHdfsPath:文件的路径
*/
public static FileStatus showFileStatus(String isHdfsPath){
FileStatus fileinfo = getFileStatus(isHdfsPath);
System.out.println(fileinfo);
System.out.println("文件大小：" + showSize(fileinfo.getLen()));
System.out.println("问价是否为目录：" + fileinfo.isDirectory());
System.out.println("文件的相同份数：" + fileinfo.getReplication()); //文件备份数最多为512个
System.out.println("文件的拆分的块的大小：" + fileinfo.getBlockSize()); 
System.out.println("文件的最后修改时间：" + showDate(fileinfo.getModificationTime()));
System.out.println("文件的操作权限：" + fileinfo.getPermission());
System.out.println("文件的拥有者：" + fileinfo.getOwner());
System.out.println("文件的所属组：" + fileinfo.getGroup());
try {
System.out.println("文件的关联：" + (fileinfo.isSymlink() ? fileinfo.getSymlink() : "没有关联文件"));
} catch (IOException e) {
e.printStackTrace();
}
return fileinfo;
}


/*转换毫秒数成日期字符串
*@param time 毫秒数
*@return 日期字符串
*/
public static String showDate(long time){
Date date = new Date(time);
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
return sdf.format(date);
}

}

hadoop2.7入门-java访问hfs文件
1、hdfs文件存储系统
2、解决hadoop集群的访问权限
2.1、关闭hadoop权限检查
2.2、使用kerberos进行权限配置
2.3、把整个hadoop的目录的权限改成777
2.4、在windows环境中添加系统环境变量hadoopUserName
3、java访问hdfs
3.1、使用URL获取文件流信息
3.2、通过hdfs的FileSystem操作文件

1、hdfs文件存储系统
使用者可以将其理解成一个文件系统，就是往里面放文件，取文件操作。

其有一个或者多个namenode，用来管理文件目录

有多个datanode用来存文件（相当于大硬盘），当datanode空间不够的时候，直接继续加datanode就可以扩展了（只要namenode的内存还能存得下目录结构）
2、解决hadoop集群的访问权限
hadoop集群默认使用linux的权限控制，windows或者java项目要操作hadoop集群的文件，则需要先解决权限问题。

这里主要有三种解决方案：
2.1、关闭hadoop权限检查
如果是测试功能用的话，最简单的办法是关闭hadoop的权限检查

修改配置文件hdfs-site.xml，增加dfs.permissions.enabled，属性设置为false

vim /usr/local/hadoop-2.7.3/etc/hadoop/hdfs-site.xml

增加：
  <property>
    <name>dfs.permissions.enabled</name>
    <value>false</value>
  </property>

保存后，重启namenode

hadoop-daemon.sh stop namenode

hadoop-daemon.sh start namenode
2.2、使用kerberos进行权限配置
此方法相当复杂，这里不研究。
2.3、把整个hadoop的目录的权限改成777
这种方法安全性太低，不推荐使用
2.4、在windows环境中添加系统环境变量hadoopUserName
3、java访问hdfs
首先建立java项目，编译环境JDK1.8

如果是传统项目，则引入hadoop对应的jar包、hdfs对应的jar包

如果是maven项目，则增加以下依赖：
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.7.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.7.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.7.3</version>
        </dependency>

java基本环境准备好了，现在可以编写java程序来测试操作hdfs文件系统：
3.1、使用URL获取文件流信息
我们先测试一下URL的使用
package com.demo.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.InputStream;
import java.net.URL;

public class HdfsTest {
    public static void main(String[] args) throws Exception{
        URL url = new URL("http://www.baidu.com");
        InputStream in = url.openStream();
        IOUtils.copyBytes(in, System.out, 4096, true);

        /*URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
        URL url = new URL("hdfs://192.168.13.135:9000/core-site.xml");
        InputStream in = url.openStream();
        IOUtils.copyBytes(in, System.out, 4096, true);

        Configuration config = new Configuration();
        config.set("fs.defaultFS","hdfs://192.168.13.135:9000");
        FileSystem fs = FileSystem.get(config);

        Boolean bl = fs.mkdirs(new Path("/test"));*/
    }
}


URL默认是http协议，直接可以访问http协议的地址；

如果我们直接访问hdfs文件系统，可以测试代码如下：
package com.demo.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.InputStream;
import java.net.URL;

public class HdfsTest {
    public static void main(String[] args) throws Exception{
        /*URL url = new URL("http://www.baidu.com");
        InputStream in = url.openStream();
        IOUtils.copyBytes(in, System.out, 4096, true);*/

        /*URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
        */
         
        URL url = new URL("hdfs://192.168.13.135:9000/core-site.xml");
        InputStream in = url.openStream();
        IOUtils.copyBytes(in, System.out, 4096, true);

        /*Configuration config = new Configuration();
        config.set("fs.defaultFS","hdfs://192.168.13.135:9000");
        FileSystem fs = FileSystem.get(config);

        Boolean bl = fs.mkdirs(new Path("/test"));*/
    }
}


执行：报错：



提示为未知协议 hdfs，我们需要将hdfs的协议注册给URL才能识别，因此加上这一句：

URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
package com.demo.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.InputStream;
import java.net.URL;

public class HdfsTest {
    public static void main(String[] args) throws Exception{
        /*URL url = new URL("http://www.baidu.com");
        InputStream in = url.openStream();
        IOUtils.copyBytes(in, System.out, 4096, true);*/

        URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());

        URL url = new URL("hdfs://192.168.13.135:9000/core-site.xml");
        InputStream in = url.openStream();
        IOUtils.copyBytes(in, System.out, 4096, true);

        /*Configuration config = new Configuration();
        config.set("fs.defaultFS","hdfs://192.168.13.135:9000");
        FileSystem fs = FileSystem.get(config);

        Boolean bl = fs.mkdirs(new Path("/test"));*/
    }
}


URL方式访问获取文件流的方式完成了，主要步骤为

①注册hdfs协议名称

②普通连接URL方式创建URL连接获取文件流

③处理获得的输入流，比如保存、获取数据、输出等
3.2、通过hdfs的FileSystem操作文件
通过这三行代码：

//这是hdfs的配置类，可以添加配置信息

Configuration config = new Configuration();

config.set(“fs.defaultFS”,“hdfs://192.168.13.135:9000”);

FileSystem fs = FileSystem.get(config);

得到hdfs的文件系统对象。

该对象可以实现：创建目录、删除文件或目录、上传文件、获取文件、判断文件或者目录是否存在

例如：
package com.demo.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;

import java.io.FileInputStream;
import java.io.InputStream;

public class HdfsTest {
    public static void main(String[] args) throws Exception{
        Configuration config = new Configuration();
        config.set("fs.defaultFS","hdfs://192.168.13.135:9000");
        FileSystem fs = FileSystem.get(config);

        Boolean bl = fs.mkdirs(new Path("/test"));
        System.out.println(bl);
        //创建目录/test2/test22，返回创建结果
        bl = fs.mkdirs(new Path("/test2/test22"));
        System.out.println(bl);
        //查看目录/test2是否存在
        bl = fs.exists(new Path("/test2"));
        System.out.println(bl);
        //上传，hdfs的目标位置
        FSDataOutputStream out = fs.create(new Path("/test2/testData.txt"));
        //上传，本地的文件
        InputStream in = new FileInputStream("D:\\worksoft\\Notepad++\\readme.txt");
        //IOUtils 可以实现上传或者下载，只要调整对应的输入和输出流即可
        IOUtils.copyBytes(in, out,4096, true);

        //列出/test2目录下的文件(夹)列表及状态
        FileStatus[] fileStatuses = fs.listStatus(new Path("/test2"));
        for (FileStatus fileStatus : fileStatuses) {
            //获取文件权限
            System.out.println(fileStatus.getPermission());
            //获取文件路径
            System.out.println(fileStatus.getPath());
            //获取文件的拷贝份数
            System.out.println(fileStatus.getReplication());
        }

        /*
         * 下载，获取对应的文件流，写入本地磁盘
         * 四个参数分别是：是否删除原文件、源文件地址、拷贝的目标文件地址、是否使用本地文件系统
        */
        fs.copyToLocalFile(false, new Path("/test2/testData.txt"), new Path("d:\\test.txt"),true);

    }
}

package hadoop;

import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URL;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;

public class Hellohdfs {
    
    private static final String PATH="hdfs://192.168.199.132:9000/test.txt";

    public static void main(String[] args) throws Exception {
        //设定开启HDFS协议
        URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());

        //开启URL
        InputStream in =  new URL(PATH).openStream();

        //读取文件
        BufferedReader reader = new BufferedReader(new InputStreamReader(in));
        String line = null;
        while((line =reader.readLine())!= null){
            System.out.println(line.toString());
        }

       
//      org.apache.hadoop.io.IOUtils.copyBytes(in, System.out, 1024,true);
    }
}



此时会出现connected refused问题这时候按以下步骤修改

把core-site.xml中的localhost 改为本机的ip地址

然后修改/etc/hosts文件

把localhost改为自己的ip地址

并添加

0.0.0.0  192.168.199.132（自己的ip地址）

摘要
通过从零开始创建一个maven工程，运用HDFS的JAVA API远程操作HDFS文件系统，通过实战提升编程能力。
1：项目创建
1：创建maven项目
打开eclipse—file—new—maven project--next     #选择工作空间，我用的默认
2：配置参数
输入Group Id和Artifact Id，这里输入com.scitc和hdfs，填写完毕后点“Finish”完成工程项目的创建。
3：检查jdk版本    #默认是javase1.5，改成自己的jdk1.8
创建完毕后检查下jdk版本是否正确，在鼠标移动到项目上点鼠标右键 -> Build Path ->Configure Build Path。
弹出一个窗口，选择Java Build Path -> JRE System Library -> Edit
弹出另一窗口，选择Workspace default JRE，Finish完成设置。
接着选择Java Compiler，将JDK版本改为1.8，点Apply后，点OK关闭窗口。
4：检查maven
依次选择window—preferences—maven设置
--选中installations：右边add
E:\java\maven\apache-maven-3.5.0   #这是我的maven的主目录
--选中user settings：右边browse如下内容
global settings：E:\java\maven\repository\settings.xml
user settings： E:\java\maven\repository\settings.xml
local repository： E:\java\maven\repository\maven_jar
#说明：如上三个大家需要根据自己安装的maven目录来选择
到此，maven项目hdfs创建结束，jdk、maven配置结束。
2：项目开发
1：编写jar包依赖pom.xml文件
4.0.0
com.scitc
hdfs
0.0.1-SNAPSHOT
jar
hdfs
http://maven.apache.org
#配置hadoop的版本号，我的是2.7.5
UTF-8
2.7.5
junit
junit
3.8.1
test
org.apache.hadoop
hadoop-mapreduce-client-common
${hadoop.version}
org.apache.hadoop
hadoop-mapreduce-client-jobclient
${hadoop.version}
provided
org.apache.hadoop
hadoop-client
${hadoop.version}
org.apache.hadoop
hadoop-yarn-common
${hadoop.version}
org.apache.hadoop
hadoop-mapreduce-client-core
${hadoop.version}
org.apache.hadoop
hadoop-hdfs
${hadoop.version}
jdk.tools
jdk.tools
1.8
system
${JAVA_HOME}/lib/tools.jar
org.apache.maven.plugins
maven-compiler-plugin
1.8
1.8
2：新建log4j.properties文件
该文件放在src/main/java代码目录的根目录下，内容如下：
log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
备注：在运行APP.java之前，需要先启动hadoop。
3：编辑App.java类
packagecom.scitc.hdfs;importjava.io.File;importjava.io.IOException;importjava.io.InputStream;importjava.security.PrivilegedExceptionAction;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FSDataOutputStream;importorg.apache.hadoop.fs.FileStatus;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IOUtils;importorg.apache.hadoop.security.UserGroupInformation;/***实现java远程操作hdfs：远程创建目录、创建文件、上传文件、下载文件、读取文件、重命名、删除文件*/
public classApp
{static Configuration conf = newConfiguration();staticFileSystem hdfs;//初始化访问hdfs的配置信息
static{
UserGroupInformation ugi= UserGroupInformation.createRemoteUser("root");try{
ugi.doAs(new PrivilegedExceptionAction() {public Void run() throwsException {
Configuration conf= newConfiguration();
conf.set("fs.defaultFS", "hdfs://192.168.56.110:9000/");//以下两行是支持 hdfs的追加功能的：hdfs.append()conf.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
conf.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
Path path= new Path("hdfs://192.168.56.110:9000/");
hdfs=FileSystem.get(path.toUri(), conf);//hdfs = path.getFileSystem(conf);//这个也可以
return null;
}
});
}catch(IOException e) {
e.printStackTrace();
}catch(InterruptedException e) {
e.printStackTrace();
}
}/*** 方法1：创建目录
*@paramdir
*@throwsIOException*/
public static void createDir(String dir) throwsIOException {//String dir = "/test3/";Path path= newPath(dir);//判断文件夹是否已存在，如果已存在就不再创建。
if(hdfs.exists(path)) {
System.out.println("文件夹 \t" + dir + "\t 已存在");return;
}//开始创建文件夹hdfs.mkdirs(path);
System.out.println("新建文件夹 \t" +dir);
}/*** 方法2：创建文件
*@throwsIOException*/
public static void createFile() throwsIOException{
String fileName= "/test/myfile.txt";
String fileContent= "this is new file";
Path dst= newPath(fileName);if(hdfs.exists(dst)) {
System.out.println("Error:文件已存在");
}else{//将文件内容转成字节数组
byte[] bytes=fileContent.getBytes();
FSDataOutputStream output=hdfs.create(dst);
output.write(bytes);
output.close();
System.out.println("创建文件\t"+fileName);
}
}/*** 方法3：copyFromLocalFile方法实现本地文件上传到hdfs上
*@paramlocalSrc
*@paramhdfsDst
*@paramfileName
*@throwsIOException*/
public static void copyFile(String localSrc, String hdfsDst, String fileName) throwsIOException {if("".equals(localSrc)){
localSrc= "E:/java/data/myfile.txt";
}if("".equals(hdfsDst)){
hdfsDst= "/test/";
}
Path src= newPath(localSrc);
Path dst= newPath(hdfsDst);//本地文件不存在
if (!(newFile(localSrc)).exists()) {
System.out.println("Error: 本地文件 \t" + localSrc + "\t 不存在。");return;
}//hdfs路径不存在
if (!hdfs.exists(dst)) {
System.out.println("Error: hdfs目录 \t" + dst.toUri() + "\t 不存在。");return;
}if("".equals(fileName)) {
fileName=src.getName();
}
String dstPath= dst.toUri() + "/" +fileName;
System.out.println(dstPath);//"/test2/myfile.txt"Path targetPath= newPath(dstPath);//判断上传的文件 hdfs的目录下是否存在
if(hdfs.exists(targetPath)) {
System.out.println("Warn: 文件 \t" + dstPath + "\t 已存在。");
}else{//本地文件上传hdfshdfs.copyFromLocalFile(src, targetPath);//遍历该目录文件FileStatus files[]=hdfs.listStatus(dst);
System.out.println("上传到 \t" + conf.get("fs.defaultFS") +hdfsDst);for(FileStatus file : files) {
System.out.println(file.getPath());
}
}
}/*** 方法4：copyToLocalFile方法实现HDFS文件下载到本地
*@throwsIllegalArgumentException
*@throwsIOException*/
public static void downloadFile() throwsIllegalArgumentException,IOException{
String hdfsDst= "/test/myfile.txt";
String localSrc= "E:/java";
Path dst= newPath(hdfsDst);
Path src= newPath(localSrc);
String localFile= localSrc + "/" + dst.getName(); //本地的路径 + hdfs下载的文件名
if(!hdfs.exists(dst.getParent())){ //如果HDFS路径不存在System.out.println("Error : HDFS路径:\t" + dst.getParent() + "\t 不存在!");return;
}if(!new File(localSrc).exists()){ //如果本地目录不存在，则创建
newFile(localSrc).mkdirs();
System.out.println("Warn : 本地目录已创建!");
}if(new File(localFile).exists()){ //如果本地文件存在System.out.println("Error : 本地文件: \t" + localFile + "\t 已存在.");return;
}if(!hdfs.exists(new Path(hdfsDst))){ //如果HDFS文件不存在System.out.println("Error : HDFS文件: \t" + hdfsDst + "\t 不存在.");
}else{//HDFS下载文件到本地hdfs.copyToLocalFile(false, dst, src, true);
System.out.println("successful ：下载成功! 请查看: \t" +localSrc);
}
}/*** 方法5：读取HDFS文件，并在本地控制台打印
*@throwsIOException*/
public static void readFile() throwsIOException {
String uri= "/test/myfile.txt";//判断文件是否存在
if (!hdfs.exists(newPath(uri))) {
System.out.println("Error;文件不存在");return;
}
InputStream in= null;try{
in= hdfs.open(newPath(uri));//复制到标准输出流IOUtils.copyBytes(in, System.out,4096, false);
}catch(Exception e) {
e.printStackTrace();
}finally{
IOUtils.closeStream(in);
}
}/*** 方法6：重命名hdfs上面的文件
*@throwsIOException*/
public static void renameFile() throwsIOException{
String oldName="/test/myfile.txt";
String newName="/test/myfile1.txt";
Path oldPath=newPath(oldName);
Path newPath=newPath(newName);if(hdfs.exists(oldPath)) {
hdfs.rename(oldPath, newPath);
System.out.println("rename success");
}else{
System.out.println("文件不存在,rename fail");
}
}/*** 方法7：给hdfs上面的文件追加内容
*@throwsIOException*/
public static void appendFile() throwsIOException{
String fileName= "/test/myfile1.txt";
String appendContent= "这是追加的内容";
Path dst= newPath(fileName);byte[] bytes =appendContent.getBytes();//如果文件不存在
if(!hdfs.exists(dst)) {
System.out.println("Error:文件不存在");return;
}
FSDataOutputStream output=hdfs.append(dst);
output.write(bytes);
output.close();
System.out.println("success:追加内容到\t"+fileName);
}/*** 方法8：删除hdfs上面的文件
*@paramfileName
*@throwsIOException*/
public static void deleteFile(String fileName) throwsIOException{if("".equals(fileName)) {
fileName= "/test/myfile1.txt";
}
Path f= newPath(fileName);boolean isExists =hdfs.exists(f);if(isExists) {boolean isDel = hdfs.delete(f,true);
System.out.println(fileName+"删除状态："+isDel);
}else{
System.out.println(fileName+"文件不存在！");
}
}public static void main( String[] args ) throwsIOException
{
System.out.println("Hello World!");//createDir("/test1");//createFile();//copyFile("E:/java/data/myfile.txt", "/test/", "myfile.txt");//downloadFile();//readFile();//renameFile();//appendFile();deleteFile("/test/myfile1.txt");
}
}