向HDFS中上传任意文本文件，如果指定的文件在HDFS中已经存在，由用户指定是追加到原有文件末尾还是覆盖原有的文件_编写一个java程序,向hdfs中上传任意文本文件,如果指定的文件在hdfs中已经存在,由

import java.io.FileInputStream;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataOutputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

public class CopyFromLocalFile {

/**

    * 判断路径是否存在

 */

public static boolean test(Configuration conf, String path) {

       try (FileSystem fs = FileSystem.get(conf)) {

           return fs.exists(new Path(path));

       } catch (IOException e) {

 

           e.printStackTrace();

           return false;

       }

 

   }

   /**

 

    * 复制文件到指定路径 若路径已存在，则进行覆盖

 

    */

 

   public static void copyFromLocalFile(Configuration conf,

           String localFilePath, String remoteFilePath) {

       Path localPath = new Path(localFilePath);

       Path remotePath = new Path(remoteFilePath);

       try (FileSystem fs = FileSystem.get(conf)) {

           /* fs.copyFromLocalFile 第一个参数表示是否删除源文件，第二个参数表示是否覆盖 */

           fs.copyFromLocalFile(false, true, localPath, remotePath);

       } catch (IOException e) {

           e.printStackTrace();

       }

   }

 

   /**

 

    * 追加文件内容

 

    */

 

   public static void appendToFile(Configuration conf, String localFilePath,

           String remoteFilePath) {

       Path remotePath = new Path(remoteFilePath);

       try (FileSystem fs = FileSystem.get(conf);

               FileInputStream in = new FileInputStream(localFilePath);) {

           FSDataOutputStream out = fs.append(remotePath);

           byte[] data = new byte[1024];

           int read = -1;

           while ((read = in.read(data)) > 0) {

               out.write(data, 0, read);

           }

           out.close();

       } catch (IOException e) {

           e.printStackTrace();

       }

   }

 

   /**

 

    * 主函数

 

    */

 

   public static void main(String[] args) {

       Configuration conf = new Configuration();

       conf.set("fs.defaultFS", "hdfs://localhost:9000");

       String localFilePath = "/usr/local/hadoop/test.txt"; // 本地路径

       String remoteFilePath = "/user/hadoop/test.txt"; // HDFS路径

       String choice = "append"; // 若文件存在则追加到文件末尾

       //String choice = "overwrite"; // 若文件存在则覆盖

       try {

           /* 判断文件是否存在 */

           boolean fileExists = false;

           if (CopyFromLocalFile.test(conf, remoteFilePath)) {

               fileExists = true;

               System.out.println(remoteFilePath + " 已存在.");

           } else {

               System.out.println(remoteFilePath + " 不存在.");

           }

           /* 进行处理 */

           if (!fileExists) { // 文件不存在，则上传

               CopyFromLocalFile.copyFromLocalFile(conf, localFilePath,

                       remoteFilePath);

               System.out.println(localFilePath + " 已上传至 " + remoteFilePath);

           } else if (choice.equals("overwrite")) { // 选择覆盖      

               CopyFromLocalFile.copyFromLocalFile(conf, localFilePath,

                       remoteFilePath);

               System.out.println(localFilePath + " 已覆盖 " + remoteFilePath);

           } else if (choice.equals("append")) { // 选择追加                

               CopyFromLocalFile.appendToFile(conf, localFilePath,

                       remoteFilePath);

               System.out.println(localFilePath + " 已追加至 " + remoteFilePath);

           }

       } catch (Exception e) {

   

           e.printStackTrace();

       }

   }

}