Java API操作 上傳文件

因為Hadoop中關於文件操作類基本上都在“org.apache.hadoop.fs”包中，這些API的主要作用主要體現在以下操作上：打開文件、讀寫文件、刪除文件。並且，Hadoop類庫中最終面向用戶提供的接口類是FileSystem，該類是一個抽象類，只能通過get方法得到。

下面，筆者就逐一的對這11個比較常用的API進行講解。

1.上傳本地文件
通過“FileSystem.copyFromLocalFile(Path src,Path dst)”可以將本地文件上傳到HDFS的指定位置上。具體代碼如下：

package HDFS.learnself;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class Upload_Files {
 public static void main(String[] args) throws IOException {
  //1.加載hdfs的配置文件
  Configuration conf=new Configuration();
  conf.set("fs.defaultFS", "hdfs://hdp02:9000");
  //2.獲取hdfs的操作對象，得到一個FileSystem對象
  FileSystem fs=FileSystem.get(conf);
  //3.創建源目的文件路徑和文件上傳操作
  Path src=new Path("D:\\hdfs.txt");
  Path dst=new Path("/");
  fs.copyFromLocalFile(src, dst);
  //4.關閉流
  fs.close();
  System.out.println("文件上傳成功！");

      }
 }

2.創建HDFS目錄
通過“fs.mkdirs”方法進行目錄的創建。具體代碼如下：

package HDFS.learnself;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class mkdir_list {
 public static void main(String[] args) throws IOException {
    //1.加載hdfs的配置信息
     Configuration conf=new Configuration();
     conf.set("fs.defaultFS", "hdfs://hdp02:9000");
    //2..獲取hdfs的操作對象
     FileSystem fs=FileSystem.get(conf);
    //3.創建目錄
     Path path=new Path("/list");
     fs.mkdirs(path);
    //4.關閉流
     fs.close();
     System.out.println("目錄創建成功!");

       }
  }

3.寫文件
通過“writeUTF()”方法可以實現對指定文件進行寫操作。具體代碼如下：

package HDFS.learnself;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Hdfs;
import org.apache.hadoop.fs.Path;

public class Write {
 public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
  //1.加載hdfs的配置文件
  Configuration conf=new Configuration();
  //2.獲取hdfs的操作對象
  FileSystem fs=FileSystem.get(new URI("hdfs://hdp02:9000"), conf, "hdp02");
  //3.文件路徑
  Path File=new Path("hdfs://hdp02:9000/test/cao.txt");
  //4.創建FSDataOutputStream對象
  FSDataOutputStream out=fs.create(File);
  //5.寫入數據
  out.writeUTF("Hello world!");
  out.close();
  System.out.println("數據寫入成功！");
    }
 }

4.讀文件
通過“ReadUTF()”方法可以實現對指定文件進行讀操作。具體代碼如下：

package HDFS.learnself;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class Read {
 public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
  //1.加載hdfs的配置文件
  Configuration conf=new Configuration();
  //2.獲取hdfs的操作對象
  FileSystem fs=FileSystem.get(new URI("hdfs://hdp01:9000"), conf, "hdp02");
  //3.讀取文件的路徑
  Path File=new Path("hdfs://hdp01:9000/test/cao.txt");
  //4.創建FSDataInputStream對象
  FSDataInputStream in=fs.open(File);
  String info = in.readUTF();
  System.out.println("數據讀取成功！");
  System.out.println(info);
  in.close();
      }
 }

5.重命名
通過“fs.rename()”方法可以實現對指定文件進行重命名。具體代碼如下：

package HDFS.learnself;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class ChangeName {
 public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
  //1.加載hdfs的配置文件
  Configuration conf=new Configuration();
  //2.獲取hdfs的操作對象
  FileSystem fs=FileSystem.get(new URI("hdfs://hdp01:9000"), conf, "hdp02");
  //3.對文件名進行操作
  Path old=new Path("/test/fur.txt");
  Path now=new Path("/test/fur01.txt");
  //4.調用hdfs的rename重命名方法，返回值為boolean類型
  boolean isRename=fs.rename(old, now);
  System.out.println("重命名成功！");
       }
 }

6.刪除文件
通過“fs.delete()”方法可以實現對指定文件進行刪除。具體代碼如下：

package HDFS.learnself;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class DeleteFile {
 public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
  //1.加載hdfs的配置文件
  Configuration conf=new Configuration();
  //2.獲取hdfs的操作對象
  FileSystem fs=FileSystem.get(new URI("hdfs://hdp01:9000"), conf, "hdp02");
  //3.將要刪除的文件路徑
  String File="/test/cao.txt";
  //4.刪除文件
  fs.delete(new Path(File), true);
  fs.close();
  System.out.println("文件刪除成功！");
      }
 }

7.刪除目錄
通過“fs.delete()”方法可以實現對目錄的（遞歸）刪除。具體代碼如下：

package HDFS.learnself;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class DeleteList {
 public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
  //1.加載hdfs的配置文件
  Configuration conf=new Configuration();
  //2.獲取hdfs的操作對象
  FileSystem fs=FileSystem.get(new URI("hdfs://hdp01:9000"), conf, "hdp02");
  //3.將要刪除的目錄路徑
  Path list=new Path("/list");
  //4.遞歸刪除
  boolean isDelete=fs.delete(list, true);
  fs.close();
  System.out.println("目錄刪除成功！");
      }
 }

8.讀取某個目錄下的所有文件
通過“FileStatus.getPath()”方法可以查看指定目錄下的所有文件。具體代碼如下：

package HDFS.learnself;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class CatFiles {
 public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
  //1.加載hdfs的配置文件
  Configuration conf=new Configuration();
  //2.獲取hdfs的操作對象
  FileSystem fs=FileSystem.get(new URI("hdfs://hdp01:9000"), conf, "hdp02");
  //3.創建要讀取的文件路徑
  Path listf=new Path("/tmp");
  //4.創建FileStatus對象，調用listStatus方法
  FileStatus stats[]=fs.listStatus(listf);
  for(int i=0;i<stats.length;i++){
   System.out.println(stats[i].getPath().toString());
  }
  fs.close();
  System.out.println("該目錄下的文件查找完畢！");
      }
}
9.查看文件是否存在
通過“FileSystem.exists(Path f)”可以查看指定文件是否存在。具體代碼如下：

package HDFS.learnself;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class ifExists {
 public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
  //1.加載hdfs的配置文件
  Configuration conf=new Configuration();
  //2.獲取hdfs的操作對象
  FileSystem fs=FileSystem.get(new URI("hdfs://hdp01:9000"), conf, "hdp02");
  //3.要查找的文件路徑
  Path File=new Path("/test/fur01.txt");
  //4.調用exists方法
  boolean isExists=fs.exists(File);
  System.out.println(isExists);
       }
  }

10.查看文件最后修改時間
通過“FileSystem.getModificationTime()”可以查看制定文件的最會修改時間（注意:這里的時間是以時間戳的形式顯示的）。具體代碼如下：

package HDFS.learnself;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class Modify_Time {
 public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
  //1.加載hdfs的配置文件
  Configuration conf=new Configuration();
  //2.獲取hdfs的操作對象
  FileSystem fs=FileSystem.get(new URI("hdfs://hdp01:9000"), conf, "hdp02");
  //3.指定文件的路徑
  Path File=new Path("/test/fur01.txt");
  //4.創建FileStatus對象，調用listStatus方法
  FileStatus filestatus=fs.getFileStatus(File);
  //5.調用getModificationTime方法
  long time = filestatus.getModificationTime();
  System.out.println(time);
  fs.close();
        }
  }

11.查看某個文件在HDFS集群中的位置
通過“FileSystem.getFileBlockLocation(FileStatus file,long start,long len)”可以查看指定文件所在位置。具體代碼如下：

package HDFS.learnself;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class WhereFile {
 public static void main(String[] args) throws IOException, InterruptedException, URISyntaxException {
  //1.加載hdfs的配置文件
  Configuration conf=new Configuration();
  //2.獲取hdfs的操作對象
  FileSystem fs=FileSystem.get(new URI("hdfs://hdp01:9000"), conf, "hdp02");
  //3.需要查找文件的路徑
  Path File=new Path("/test/fur01.txt");
  //4.創建FileStatus對象，調用getFileStatus方法
  FileStatus filestatus=fs.getFileStatus(File);
  //5.創建一個BlockLocation數組對象，調用getFileBlockLocations方法
  BlockLocation [] blkLocations=fs.getFileBlockLocations(filestatus, 0, filestatus.getLen());
  int blockLen=blkLocations.length;
  for(int i=0;i<blockLen;i++){
   String [] hosts=blkLocations[i].getHosts();
   System.out.println("blockID:"+i+"\n"+"location:"+hosts[0]);
      }
 }


說明：
1.在最開始的兩個API(文件上傳和創建目錄)的時候，采用的是conf.set(）的方法，但在后面的API操作中，統一的改為了FileSystem.get(uri，conf，user)的形式進行HDFS的配置的。主要是因為按照第一種方式，經常容易出現權限不足的異常，而采用后面的格式，一來防止出現權限異常問題，而來也改進了代碼冗余問題。（推薦讀者使用后面的編寫方式）
2.在FileSystem.get(uri，conf，user)中指定URI的時候，既寫了“hdfs://hdp02:9000”,也寫了“hdfs://hdp01:9000”。主要是因為搭建的是高可用集群，有兩個主節點（hdp01和hdp02），而不巧的是，在編程的過程中，開始的hdp02主節點宕掉了，所以后面處於Active狀態的是hdp01主節點。（請讀者根據自己的實際情況編寫）
3.在FileSystem.get(uri，conf，user)中user是指集群登錄的用戶名。

Java API操作 上傳文件

package com.hdfs;



import java.net.URI;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.Before;
import org.junit.Test;



public class HdfsTest2 {
    
    //獲取 FileSystem 對象
    private FileSystem fs=null;
    @SuppressWarnings("unused")
    private List<String> hdfsPathsLists;
    
    
   @Before
   
 
    public void init() throws Exception {
        Configuration con = new Configuration();
        con.setBoolean("dfs.support.append",true);
        fs = FileSystem.get(new URI("hdfs://192.168.119.128:9000"),
                con,"root");
        
    }
    @Test
   public void testUploadFile() throws Exception{
        String src = "/home/cai/123.cfg";
        String hdfsDst="/javaApi/mk/dir1";
        fs.copyFromLocalFile(new Path(src),new Path(hdfsDst));
        System.out.println("upload success");
    }
}