java api操作HDFS

使用是使用maven的话导入如下依赖即可否则需要在解压好的hadoop文件夹下找到common文件夹和hdfs文件夹下的jar包

<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.6.1</version>
</dependency>

要进行操作，主要得先拿到客户端对象

public class HdfsClient {
Configuration conf = null;
FileSystem fileSystem = null;
@Before
public void init() throws Exception{
conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://192.168.25.127:9000");
fileSystem = FileSystem.get(conf);
}
}

解释：我们的操作目标是HDFS，所以获取到的fs对象应该是DistributedFileSystem的实例；get方法是从何处判断具体实例化那种客户端类呢？

——从conf中的一个参数 fs.defaultFS的配置值判断；

如果我们的代码中没有指定fs.defaultFS，并且工程classpath下也没有给定相应的配置，conf中的默认值就来自于hadoop的jar包中的core-default.xml，默认值为： file:///，则获取的将不是一个DistributedFileSystem的实例，而是一个本地文件系统的客户端对象。那么本地（一般我们使用的就是windows）需要安装配置hadoop，还要编译，配置环境变量，会比较麻烦，所以我们连接到linux。

关于conf.set(name,value)是设置配置参数的，也可以在classpath下加入配置文件hdfs-default.xml进行配置，或者使用jar包中的配置（默认的），优先级是由高到低。

比如con.set(“dfs.replication”,4),配置文件中配置的副本为2，包中默认的是3，最后副本数量是4。

1、测试文件上传

/**
* 测试上传
* d:/mylog.log传到hdfs路径/mylog.log.copy
* @throws Exception
*/
@Test
public void testUpload() throws Exception{
fileSystem.copyFromLocalFile(new Path("d:/mylog.log"), new Path("/mylog.log.copy"));
fileSystem.close();
}

页面查看效果



2、测试下载文件，将刚上传的/mylog.log.copy下载到本地指定位置

/**
* 测试下载
* 第一个参数表示是否删除源文件，即：剪切+粘贴
* 最后一个参数表示是否使用本地文件系统，不使用的话会使用io，如果本地没配置hadoop的话会出现空指针异常。
* @throws Exception
*/
@Test
public void testdownLoad() throws Exception{
fileSystem.copyToLocalFile(true, new Path("/mylog.log.copy"), new Path("d:/zz.log"),
true);
fileSystem.close();
}

3、获取配置参数

/**
* 获取配置参数:获取的是jar包中配置的，服务端配置的是不起作用的
* 但是可以使用配置文件或者用cong.set()来指定
* 比如副本数量 dfs.replication,3
* @throws Exception
*/
@Test
public void testConfiguration() throws Exception{
Iterator<Entry<String, String>> it = conf.iterator();
while (it.hasNext()){
Entry<String, String> entry = it.next();
System.out.println(entry.getKey()+","+entry.getValue());
}
fileSystem.close();
}

4、测试创建文件夹，可以创建多层

/**
* 测试创建文件夹 可以是多层
* @throws Exception
*/
@Test
public void testMkdir() throws Exception{
boolean b = fileSystem.mkdirs(new Path("/djhot/cls"));
System.out.println("文件夹是否创建成功:" + b);
fileSystem.close();
}

5、测试删除文件或文件夹

/**
* 测试删除文件夹或文件，第二个参数用true则表示递归删除
* @throws Exception
*/
@Test
public void testDelete() throws Exception{
boolean b = fileSystem.delete(new Path("/djhot"),true);
System.out.println("文件夹是否删除:" + b);
boolean c = fileSystem.delete(new Path("/cenos-6.5-hadoop-2.6.4.tar.gz"),true);
System.out.println("文件是否删除：" + c);
fileSystem.close();
}

5、列出所有文件以及相关信息

：/wordcount/output/a.txt,/wordcount/output/b.txt,/wordcount/input/a.txt,/wordcount/input/b.txt,

/**
* 列出指定文件夹下的所有文件，第二个参数true表示递归列出
* 每次拿到的只有一个文件,如果不使用迭代器一次拿太多内存吃不消
* @throws Exception
*/
@Test
public void testListfiles() throws Exception{
RemoteIterator<LocatedFileStatus> files = fileSystem.listFiles(new Path("/wordcount"), true);//true表示递归
while(files.hasNext()){
LocatedFileStatus fileStatus = files.next();
System.out.println("blockSize"+fileStatus.getBlockSize());
System.out.println("owner:"+fileStatus.getOwner());
System.out.println("replication:"+fileStatus.getReplication());
//文件路径
System.out.println("path:"+fileStatus.getPath());
//文件名
System.out.println("name:"+fileStatus.getPath().getName());
//关于block块的一些信息
BlockLocation[] blockLocations = fileStatus.getBlockLocations();
for (BlockLocation block : blockLocations) {
System.out.println("块大小:"+block.getLength());
System.out.println("偏移量:"+block.getOffset());
String[] hosts = block.getHosts();
for (String host : hosts) {
System.out.println("所在datanode:"+host);
}
}
System.out.println("------------");
}
fileSystem.close();
}

输出：

blockSize134217728
owner:root
replication:2
path:hdfs://192.168.25.127:9000/wordcount/input/a.txt
name:a.txt
块大小:71
偏移量:0
所在datanode:mini2
所在datanode:mini3
------------
blockSize134217728
owner:root
replication:2
path:hdfs://192.168.25.127:9000/wordcount/input/b.tx
name:b.tx
块大小:71
偏移量:0
所在datanode:mini2
所在datanode:mini3
------------
blockSize134217728
owner:root
replication:2
path:hdfs://192.168.25.127:9000/wordcount/output/_SUCCESS
name:_SUCCESS
------------
blockSize134217728
owner:root
replication:2
path:hdfs://192.168.25.127:9000/wordcount/output/part-r-00000
name:part-r-00000
块大小:75
偏移量:0
所在datanode:mini2
所在datanode:mini3
------------

6、列出指定目录下的文件或文件夹，不会递归

/**
* 列出指定目录下的文件夹或文件  并不会递归
* @throws Exception
*/
@Test
public void testListStatus() throws Exception{
FileStatus[] listStatus = fileSystem.listStatus(new Path("/wordcount"));
for (FileStatus fileStatus : listStatus) {
System.out.println("path:"+fileStatus.getPath());
System.out.println("name:"+fileStatus.getPath().getName());
}
System.out.println("---------------------");
FileStatus[] listStatus2 = fileSystem.listStatus(new Path("/wordcount/input"));
for (FileStatus fileStatus : listStatus2) {
System.out.println("path:"+fileStatus.getPath());
System.out.println("name:"+fileStatus.getPath().getName());
}
}

输出

path:hdfs://192.168.25.127:9000/wordcount/input
name:input
path:hdfs://192.168.25.127:9000/wordcount/output
name:output
---------------------
path:hdfs://192.168.25.127:9000/wordcount/input/a.txt
name:a.txt
path:hdfs://192.168.25.127:9000/wordcount/input/b.tx
name:b.tx

7 、使用流进行文件读写

/**
* 用流的方式来操作hdfs上的文件，可以实现读取指定偏移量范围的数据
* @author 12706
*
*/
public class HdfsStreamAccess {
Configuration conf=null;
FileSystem fs=null;
@Before
public void init() throws Exception{
conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://192.168.25.127:9000");
fs = FileSystem.get(conf);
}
/**
* 通过流，将本地文件D:/liushishi.love写到hdfs下的/liushishi.love
* @throws Exception
*/
@Test
public void testUpload() throws Exception{
FSDataOutputStream outputStream = fs.create(new Path("/liushishi.love"), true);//有就覆盖
FileInputStream inputStream = new FileInputStream("D:/liushishi.love");
IOUtils.copy(inputStream, outputStream);
}
/**
* 通过流，将hdfs下的/liushishi.love文件，写到本地文件D:/liushishi.love2
* @throws Exception
*/
@Test
public void testDownload() throws Exception{
FSDataInputStream inputStream = fs.open(new Path("/liushishi.love"));
FileOutputStream outputStream = new FileOutputStream("D:/liushishi.love2");
IOUtils.copy(inputStream, outputStream);
}
/**
* 指定位置开始写
* @throws Exception
*/
@Test
public void testRandomAccess() throws Exception{
FSDataInputStream inputStream = fs.open(new Path("/liushishi.love"));
//FileInoutStream是没有这个方法的,定位到第12个字节处
inputStream.seek(12);
FileOutputStream outputStream = new FileOutputStream("D:/liushishi.love2");
//从第12个字节写到文件末
IOUtils.copy(inputStream, outputStream);
}
/**
* 将hdfs指定文件内容输出到控制台
* @throws Exception
*/
@Test
public void testCat() throws Exception{
FSDataInputStream inputStream = fs.open(new Path("/liushishi.love"));
IOUtils.copy(inputStream, System.out);
}