使用java api操作Hadoop文件

<1>. 文件操作?

1.1 上传本地文件到文件系统

?

?? ??/*
?????*?upload?the?local?file?to?the?hds?
?????*?notice?that?the?path?is?full?like?/tmp/test.c
?????*/
????public?static?void?uploadLocalFile2HDFS(String?s,?String?d)?
????????throws?IOException
????{
????????Configuration?config?=?new?Configuration();
????????FileSystem?hdfs?=?FileSystem.get(config);
????????
????????Path?src?=?new?Path(s);
????????Path?dst?=?new?Path(d);
????????
????????hdfs.copyFromLocalFile(src,?dst);
????????
????????hdfs.close();
????}

?

1.2 创建新文件，并写入

?? ??/*
?? ? *?create?a?new?file?in?the?hdfs.
?????*?notice?that?the?toCreateFilePath?is?the?full?path
?????*?and?write?the?content?to?the?hdfs?file.
?????*/
????public?static?void?createNewHDFSFile(String?toCreateFilePath,?String?content)?throws?IOException
????{
????????Configuration?config?=?new?Configuration();
????????FileSystem?hdfs?=?FileSystem.get(config);
????????
????????FSDataOutputStream?os?=?hdfs.create(new?Path(toCreateFilePath));

????????os.write(content.getBytes("UTF-8"));
????????
????????os.close();
????????
????????hdfs.close();
????}

1.3 删除文件

?? ?/*
?? ? *?delete?the?hdfs?file?
?????*?notice?that?the?dst?is?the?full?path?name
?????*/
????public?static?boolean?deleteHDFSFile(String?dst)?throws?IOException
????{
????????Configuration?config?=?new?Configuration();
????????FileSystem?hdfs?=?FileSystem.get(config);
????????
????????Path?path?=?new?Path(dst);
????????boolean?isDeleted?=?hdfs.delete(path);
????????
????????hdfs.close();
????????
????????return?isDeleted;
????}
?? ?

1.4 ?读取文件

?? ??/**?read?the?hdfs?file?content
?????*?notice?that?the?dst?is?the?full?path?name
?????*/
????public?static?byte[]?readHDFSFile(String?dst)?throws?Exception
????{
????????Configuration?conf?=?new?Configuration();
????????FileSystem?fs?=?FileSystem.get(conf);
????????
????????//?check?if?the?file?exists
????????Path?path?=?new?Path(dst);
????????if?(?fs.exists(path)?)
????????{
????????????FSDataInputStream?is?=?fs.open(path);
????????????//?get?the?file?info?to?create?the?buffer
????????????FileStatus?stat?=?fs.getFileStatus(path);
????????????
????????????//?create?the?buffer
????????????byte[]?buffer?=?new?byte[Integer.parseInt(String.valueOf(stat.getLen()))];
????????????is.readFully(0,?buffer);
????????????
????????????is.close();
????????????fs.close();
????????????
????????????return?buffer;
????????}
????????else
????????{
????????????throw?new?Exception("the?file?is?not?found?.");
????????}
????}

<2>. 目录操作

2.1 创建目录

?? ???/**?make?a?new?dir?in?the?hdfs
?????*?
?????*?the?dir?may?like?'/tmp/testdir'
?????*/
????public?static?void?mkdir(String?dir)?throws?IOException
????{
????????Configuration?conf?=??new?Configuration();
????????FileSystem?fs?=?FileSystem.get(conf);
????????
????????fs.mkdirs(new?Path(dir));
????????
????????fs.close();
????}

2.2 删除目录????

?? ???/**?delete?a?dir?in?the?hdfs
?????*?
?????*?dir?may?like?'/tmp/testdir'
?????*/
????public?static?void?deleteDir(String?dir)?throws?IOException
????{
????????Configuration?conf?=?new?Configuration();
????????FileSystem?fs?=?FileSystem.get(conf);
????????
????????fs.delete(new?Path(dir));
????????
????????fs.close();
????}

2.3 读取某个目录下的所有文件?

?? ?public?static?void?listAll(String?dir)?throws?IOException
?? ?{
????????Configuration?conf?=?new?Configuration();
????????FileSystem?fs?=?FileSystem.get(conf);
????????
????????FileStatus[]?stats?=?fs.listStatus(new?Path(dir));
????????
????????for(int?i?=?0;?i?<?stats.length;?++i)
????????{
????????????if?(stats[i].isFile())
????????????{
????????????????//?regular?file
????????????????System.out.println(stats[i].getPath().toString());
????????????}
????????????else?if?(stats[i].isDirectory())
????????????{
????????????????//?dir
????????????????System.out.println(stats[i].getPath().toString());
????????????}
????????????else?if(stats[i].isSymlink())
????????????{
????????????????//?is?s?symlink?in?linux
????????????????System.out.println(stats[i].getPath().toString());
????????????}
?????????????????
????????}
????????fs.close();
????}