HDFS的shell和API操作

1. HDFS的shell操作

hadoop  version     //查看版本
hadoop   fs   -appendToFile  src(Linux中的文件) dest(hdfs目錄下的文件)  //追加
hadoop   fs   -cat file(hdfs目錄下的文件)   //查看文件內容
Hadoop   fs   -tail file(hdfs目錄下的文件)   //查看文件末尾1kb的數據
hadoop   fs  -checksum  file(hdfs目錄下的文件)  //校驗當前文件是否正確
hadoop   fs  -copyFromLocal src dest  //從本地復制文件到HDFS
hadoop   fs  -copyToLocal   dest src  //從hdfs復制文件到本地
hadoop   fs  -count path    //對目錄內容做計數
hadoop   fs  -find  path     //查看某個hdfs目錄中是否相應的文件
hadoop   fs  -getmerge   src dest  //合并下載
hadoop   fs  -ls   path     //列表
hadoop   fs  -put  file(本地)  dest（hdfs）   //將本地文件上傳到hdfs
hadoop   fs  -setrep  num  file   //設置hdfs系統中的某個文件的副本數(只能是hdfs中已上傳的文件)
hadoop   fs  -setrep -R num file //設置hdfs系統中的目錄下的所有文件的副本數
hadoop   fs  -text 壓縮文件   //查看hdfs中的壓縮文件
hadoop   fs  -truncate   dir  //清空hdfs目錄
hdfs  getconf  -confkey  [key](配置文件的name)  //查看相應的配置的value

2. HDFS的API操作

1）eclipse中的hdfs的開發環境搭建：

• 下載一個新版的eclipse，將自己集成的hadoop-eclipse-plugin-2.7.4.jar放入eclipse的安裝目錄中的…\eclipse-hadoop\plugins中。
  插件下載地址：http://down.51cto.com/data/2457877
  然后打開eclipse：
  
  
  能查看到以上的兩個組件，表示eclipse環境配置成功。
• 配置eclipse插件連接集群
  
  配置成功之后，在DFS Locations中就會出現hdfs的文件目錄
  
• window平臺使用hadoop：
  • 在windows下配置hadoop的環境變量，將hadoop的解壓包放置一個位置，然后加入：HADOOP_HOME:../hadoop2.7.4/，在path中%HADOOP_HOME:%/bin、
    hadoop.dll,winutils.exe下載地址：http://down.51cto.com/data/2457878
  • 將hadoop.dll 放入c:\windows\system32中
  • 將winutils.exe，放入hadoop 的bin目錄下
  • 將eclipse集成hadoop
    

2）添加hadoop依賴包的三種方式：

• 在項目中創建一個文件夾，然后將所有需要依賴的jar拷貝到這個目錄，右擊add build path 將jar導入項目
• 創建一個用戶的library，在本地將相應的jar包導入到項目中：
  
  注意這里的兩個lib中所有jar包都要添加
• 使用maven導入jar包（推薦）

  3）hdfs編程中的重要的 兩個對象：

  ??Configuration類的介紹：
  ??Configuration類是加載hadoop中的相應的配置文件

  Configuration conf=new Configuration(true);
  如果在這里沒有在source folder中放入相應的配置文件：
  core-site.xml、hdfs-site.xml、yarn-site.xml、mapreduce-site.xml…
  那么Configuration類會自動的加載jar包中的配置文件：
  hadoop-hdfs-2.7.6.jar/hdfs-default.xml
  只有在sourcefolder中放入相應的配置時，才能夠加載相應的配置，但是配置文件的名稱必須是site 或者default的文件，才能夠正確加載

  ??使用相應的方法加載配置文件：

  Configuration conf=new Configuration();
  conf.addResource(""); //這例實參的內容為配置文件的權限定名稱

  ??FileSystem類的介紹：
  獲取FileSystem對象：

  //以這種方式獲取的fs對象是本地的文件系統的對象，如果在windows下就是windows的對象
  Configuration conf=new Configuration();
  FileSystem fs=FileSystem.get(conf);
  //能加載自己集群的文件系統對象
  conf.set("fs.defaultFS", "hdfs://hadoop01:9000");

  解決編程時的權限問題：
  ?由于hdfs文件系統是基于用戶的，在Windows下eclipse中操作hdfs默認的用戶是Windows的用戶，在操作hdfs文件系統時沒有權限的，所以在做讀寫操作時，可能會報出：
  
  解決方案：
  (1)在運行時使用run configuration,配置相應的參數：-DHADOOP_USER_NAME=hadoop(指定運行用戶)
  (2)在代碼中指定用戶：FileSystem.get(newURI("hdfs://hadoop01:9000"),conf,"hadoop");
  (3)在代碼中指定jvm運行時使用的用戶System.setProperty("HADOOP_USER_NAME", "hadoop"); 這里需要使用run configuration運行
  (4)在Windows中添加一個hadoop用戶，不建議使用
  如果以上方式仍然出現問題，那么就需要在window的path中配置一個環境變量：
  DHADOOP_USER_NAME = hadoop 即可。

4）hdfs的實戰編程：

public class HDFSApp {

    //文件系統
    FileSystem fileSystem = null;
    //配置類
    Configuration configuration = null;

    @Before
    public void setup() {
        configuration = new Configuration();
        configuration.set("fs.defautlFS", "hdfs://zzy:9000");
        configuration.addResource("core-site.xml");
        configuration.addResource("hdfs-site.xml");
        try {
            fileSystem = FileSystem.get(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //創建目錄
    @Test
    public void mkdir() {
        try {
            System.setProperty("HADOOP_USER_NAME", "hadoop");
            boolean isMkdirs = fileSystem.mkdirs(new Path("/user/hadoop/test"));
            if (isMkdirs) {
                System.out.println("創建成功！！");
            } else {
                System.out.println("創建失敗！！");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //刪除目錄
    @Test
    public void deletedir() {
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        try {
            fileSystem.delete(new Path("/每日任務.txt"), true);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //將本地文件copy到hdfs中
    @Test
    public void CopyFromeLocal() {
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        Path  src=new Path("C:\\Users\\aura-bd\\Desktop\\每日任務.txt");
        Path dest=new Path("/");
        try {
            fileSystem.copyFromLocalFile(src,dest);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //將hdfs文件copy到本地
    @Test public void CopyToLocal(){
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        Path  src=new Path("C:\\Users\\aura-bd\\Desktop\\");
        Path dest=new Path("/user/hive/warehouse/test.db/pokes/data.txt");
        try {
            fileSystem.copyToLocalFile(dest,src);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //顯示目錄下的文件夾信息
    @Test
    public void FSListFile(){
        try {
            RemoteIterator<LocatedFileStatus> filelist = fileSystem.listFiles(new Path("/user/hive/warehouse/test.db"), true);
            while(filelist.hasNext()){
                LocatedFileStatus fileStatus = filelist.next();
                System.out.println(fileStatus.getPath());
                System.out.println(fileStatus.getGroup());
                System.out.println(fileStatus.getPath().getName());
                System.out.println(fileStatus.getReplication());

                BlockLocation[] blockLocations = fileStatus.getBlockLocations();
                for(BlockLocation block :blockLocations){
                    System.out.println(block.getHosts().toString());
                    System.out.println(block.getNames());
                    System.out.println(block.getOffset());
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //顯示文件夾以及文件信息
    @Test
    public void ListFiles(){
        try {
            FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/"));
            for(FileStatus file:fileStatuses){
                if(file.isDirectory()){
                    System.out.println("directory:"+file.getPath().getName());
                }else{
                    System.out.println("file:"+file.getPath().getName());
                }
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //下載文件
    @Test
    public void DownLoadFileToLocal(){
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        try {
            FSDataInputStream open = fileSystem.open(new
                    Path("/user/hive/warehouse/test.db/pokes/data.txt"));
            OutputStream out=new FileOutputStream(new File("D:\\data.txt"));
            IOUtils.copyBytes(open,out,1024);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //上傳文件
    @Test
    public void upLoadFileToLocal(){
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        try {
            FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/a.txt"));
            InputStream in=new FileInputStream("D:\\data.txt");
            IOUtils.copyBytes(in,fsDataOutputStream,4096);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

3. HDFS的經典案例：

  在hadoop中MR編程計算框架中，都是計算向數據移動，那么如何獲取一個文件的所有block信息，指定block塊進行數據讀取的呢？
代碼實現：

public class RamdomRead {
    private static Configuration conf;
    private static FileSystem fs;
    static {
        conf=new Configuration(true);
        conf.set("fs.defalutFS","hdfs://zzy:9000");
        conf.addResource("core-site.xml");
        conf.addResource("hdfs-site.xml");
        try {
            fs=FileSystem.get(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) throws IOException {
        System.setProperty("HADOOP_USER_NAME","hadoop");
        Path file=new Path("/data/2018-8-8/access.log");
        FSDataInputStream open = fs.open(file);
        //拿到文件信息
        try {
             FileStatus[] fileStatuses = fs.listStatus(file);
            // 獲取這個文件的所有 block 的信息
            BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(fileStatuses[0], 0L, fileStatuses[0].getLen());
            // 第一個 block 的長度
            BlockLocation fileBlockLocation = fileBlockLocations[0];
            //第一個塊的長度
            long length = fileBlockLocation.getLength();
            //第一個 block 的起始偏移量
            long offset = fileBlockLocation.getOffset();

            //讀取數據
            byte flush[]=new byte[4096];
            FileOutputStream os = new FileOutputStream(new File("d:/block0"));
            while(open.read(offset,flush,0,4096)!=-1){
                os.write(flush);
                offset+=4096;
                if(offset>length){
                    return ;
                }
            }
            os.flush();
            os.close();
            fs.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}