java操作hdfs（第2篇）

摘要

通過從零開始創建一個maven工程，運用HDFS的JAVA API遠程操作HDFS文件系統，通過實戰提升編程能力。

Hadoop java api文檔地址：http://hadoop.apache.org/docs/current/api/index.html

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文件

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.scitc</groupId>
  <artifactId>hdfs</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>hdfs</name>
  <url>http://maven.apache.org</url>

#配置hadoop的版本號，我的是2.7.5
  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <hadoop.version>2.7.5</hadoop.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
    
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-mapreduce-client-common</artifactId>
      <version>${hadoop.version}</version>
    </dependency>


    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
      <version>${hadoop.version}</version>
      <scope>provided</scope>
    </dependency>


    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-client</artifactId>
      <version>${hadoop.version}</version>
    </dependency>
    
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-yarn-common</artifactId>
      <version>${hadoop.version}</version>
    </dependency>
     
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-mapreduce-client-core</artifactId>
      <version>${hadoop.version}</version>
    </dependency> 
    
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-hdfs</artifactId>
      <version>${hadoop.version}</version>
    </dependency>
    
    <dependency>  
      <groupId>jdk.tools</groupId>  
      <artifactId>jdk.tools</artifactId>  
      <version>1.8</version>  
      <scope>system</scope>  
      <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>  
   </dependency>
   
  </dependencies>
  
  <build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
            <source>1.8</source>
            <target>1.8</target>
            </configuration>
    </plugin>
    </plugins>
</build>
 
</project>

 

 

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類

package com.scitc.hdfs;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.security.PrivilegedExceptionAction;
import org.apache.hadoop.conf.Configuration;
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.security.UserGroupInformation;

/**
*實現java遠程操作hdfs：遠程創建目錄、創建文件、上傳文件、下載文件、讀取文件、重命名、刪除文件
*/
public class App
{
   static Configuration conf = new Configuration();
   static FileSystem hdfs;

   //初始化訪問hdfs的配置信息
   static {
      UserGroupInformation ugi = UserGroupInformation.createRemoteUser("root");
           try {
               ugi.doAs(new PrivilegedExceptionAction<Void>() {
                   public Void run() throws Exception {
                       Configuration conf = new Configuration();
                       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：創建目錄
    * @param dir
    * @throws IOException
    */
   public static void createDir(String dir) throws IOException {
        //String dir = "/test3/";
        Path path = new Path(dir);
        //判斷文件夾是否已存在，如果已存在就不再創建。
        if (hdfs.exists(path)) {
            System.out.println("文件夾 \t" + dir + "\t 已存在");
            return;
        }
        //開始創建文件夾
        hdfs.mkdirs(path);
        System.out.println("新建文件夾 \t" + dir);
}

   /**
    * 方法2：創建文件
    * @throws IOException
    */
      public static void createFile() throws IOException{
         String fileName = "/test/myfile.txt";
         String fileContent = "this is new file";
         Path dst = new Path(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上
       * @param localSrc
       * @param hdfsDst
       * @param fileName
       * @throws IOException
       */
      public static void copyFile(String localSrc, String hdfsDst, String fileName) throws IOException {
           if("".equals(localSrc)){
             localSrc = "E:/java/data/myfile.txt";
           }
           if("".equals(hdfsDst)){
             hdfsDst = "/test/";
           }
           Path src = new Path(localSrc);
           Path dst = new Path(hdfsDst);
           //本地文件不存在
           if (!(new File(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 = new Path(dstPath);
           //判斷上傳的文件 hdfs的目錄下是否存在
           if (hdfs.exists(targetPath)) {
               System.out.println("Warn: 文件 \t" + dstPath + "\t 已存在。");
           }else{
               //本地文件上傳hdfs
               hdfs.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文件下載到本地
         * @throws IllegalArgumentException
         * @throws IOException
         */
      public static void downloadFile() throws IllegalArgumentException,IOException{
           String hdfsDst = "/test/myfile.txt";
           String localSrc = "E:/java";
           Path dst = new Path(hdfsDst);
           Path src = new Path(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()){ //如果本地目錄不存在，則創建
               new File(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文件，並在本地控制台打印
       * @throws IOException
       */
      public static void readFile() throws IOException {
         String uri = "/test/myfile.txt";
         // 判斷文件是否存在
         if (!hdfs.exists(new Path(uri))) {
            System.out.println("Error;文件不存在");
            return;
         }
         InputStream in = null;
         try {
            in = hdfs.open(new Path(uri));
            // 復制到標准輸出流
            IOUtils.copyBytes(in, System.out, 4096, false);
         } catch (Exception e) {
            e.printStackTrace();
         } finally {
            IOUtils.closeStream(in);
         }
      }

      /**
       * 方法6：重命名hdfs上面的文件
       * @throws IOException
       */
      public static void renameFile() throws IOException{
         String oldName="/test/myfile.txt";
         String newName="/test/myfile1.txt";
         Path oldPath=new Path(oldName);
         Path newPath=new Path(newName);
         if(hdfs.exists(oldPath)) {
            hdfs.rename(oldPath, newPath);
            System.out.println("rename success");
         }else {
            System.out.println("文件不存在,rename fail");
         }
      }

      /**
       * 方法7：給hdfs上面的文件追加內容
       * @throws IOException
       */
      public static void appendFile() throws IOException{
         String fileName = "/test/myfile1.txt";
         String appendContent = "這是追加的內容";
         Path dst = new Path(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上面的文件
       * @param fileName
       * @throws IOException
       */
      public static void deleteFile(String fileName) throws IOException{
         if("".equals(fileName)) {
            fileName = "/test/myfile1.txt";
         }
         Path f = new Path(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 ) throws IOException
    {
        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");

    }

}