【小白视角】大数据基础实践(三)HDFS的简单基本操作

hdfs dfs -mkdir /目录

hdfs dfs -ls

hdfs dfs -rm -r /目录

hdfs dfs -put 源路径 目的路径

hdfs dfs -cat 文件路径

hdfs dfs -get 源文件地址 下载到的路径

hdfs dfs -mv 文件路径 目的目录

hdfs dfs -cp 文件路径 目的目录

hdfs dfs -touch 文件路径

hdfs dfs -appendToFile 追加的内容文件路径 被追加的文件路径

hdfs dfs -text 文件路径

hdfs dfs -rm 文件路径

import java.io.*;
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 HDFSmkdir {
    public static void appendToFile(Configuration conf, String localFilePath, String remoteFilePath) {
        Path remotePath = new Path(remoteFilePath); //传入hdfs文件路径
        try( 
        FileSystem fs = FileSystem.get(conf); 
        FileInputStream in = new FileInputStream(localFilePath);){
        //new一个文件的输入流的对象，并加入需要add的文件路径
            FSDataOutputStream out = fs.append(remotePath);  // 将hdfs中的文件读入
            byte[] data = new byte[1024]; 
            /*
            从输出流中读取一定数量的字节，并将其存储在缓冲区数组data中。返回：
          读入缓冲区的总字节数；
          如果因为已经到达流末尾而不再有数据可用，则返回 -1。
    */
            int read = -1;
            while((read = in.read(data)) > 0) {
                out.write(data,0,read); // 添加hdfs文件之中
            }
            out.close(); // 关闭hdfs的文件
        }catch (IOException e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) {
        new Mkdir(); //主函数new一个Mkdir对象
    }
}
class Mkdir{
    public Mkdir(){
        String filename = "/user/hadoop/input/hdfs-site.xml";  //需要添加本地内容的文件
        Configuration conf = new Configuration(); // new一个配置文件的对象
        conf.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
     /*
     在进行pipeline写数据（上传数据的方式）时，
     如果DN或者磁盘故障，客户端将尝试移除失败的DN，然后写到剩下的磁盘。
     一个结果是，pipeline中的DN减少了。这个特性是添加新的DN到pipeline。
     这是一个站点范围的选项。
     */
        conf.set("dfs.client.block.write.replace-datanode-on-failure.policy","NEVER");
    /*
    此属性仅在dfs.client.block.write.replace-datanode-on-failure.enable设置为true时有效。
  ALWAYS: 总是添加新的DN
  NEVER: 从不添加新的DN
  DEFAULT: 设r是副本数，n是要写的DN数。在r>=3并且floor(r/2)>=n或者r>n(前提是文件是 hflushed/appended)时添加新的DN。
  */
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
    /*
    默认文件系统的名称。URI形式。uri's的scheme需要由(fs.SCHEME.impl)指定文件系统实现类。 uri's的authority部分用来指定host, port等。默认是本地文件系统。
  HA方式，这里设置服务名，例如：hdfs://111111
  HDFS的客户端访问HDFS需要此参数。
    伪分布是这样的，如果是全分布的话，应该是你的映射名字不是localhost了。应该。 
    */
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        try {
            FileSystem fs = FileSystem.get(conf);
            Path path = new Path(filename);
            if(fs.exists(path)) {
                System.out.println("File \""+ filename +"\" Exist!");
                // 文件存在，追加内容
                try {
                    String localFilePath = "/home/zqc/add.txt"; 
                    //需要追加的文件内容的路径
                    HDFSmkdir.appendToFile(conf, localFilePath, filename);
                    System.out.println("File Written Successfully!");
                }catch(Exception e) {
                    e.printStackTrace();
                }
            }
            else {
                System.out.println("File \""+ filename +"\" Not Exist!");
                fs.createNewFile(path);  // 文件不存在，创建文件
                System.out.println("File Created Successfully!");
            }
        }catch(IOException e)
        {
            e.printStackTrace();
        }
    }
}

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Scanner;
public class HDFSFileWritten {
    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) {
        new HDFSWrite();
    }
}
class HDFSWrite {
    public HDFSWrite() {
        System.out.println("Please Choose The Way To Write");
        System.out.println("1.Written On Real Time");
        System.out.println("2.Written By Local File");
        Scanner str = new Scanner(System.in);
        String s = str.nextLine();
        String filename = "/user/hadoop/input/hdfs-site.xml";
        Path writtenpath = new Path(filename);
        Configuration conf = new Configuration();
        //配置信息都和上面的一样的！
        conf.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
        conf.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        if (s.equals("1")) { //如果选择了1的话，就是自己输入
            System.out.println("Please Input The Content You Want To Append：");
            Scanner scanner = new Scanner(System.in);
            String string;
            try { //一个死循环
                while (true) {
                    FileSystem fs = FileSystem.get(conf);
                    FSDataOutputStream fsdataoutput = fs.append(writtenpath);
                    string = scanner.nextLine();
                    if (string.equals("exit")) {  //如果输入是exit就退出
                        System.out.println("APPEND SUCCESSFULLY");
                        break;
                    }
                    try { // 不然的话就是输入输入文件的内容
                        fsdataoutput.write(string.getBytes());
                        fsdataoutput.flush();
                        fsdataoutput.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        if (s.equals("2")) {  //如果是2的话就直接添加
            try {
                String localFilePath = "/home/zqc/add.txt";
                HDFSFileWritten.appendToFile(conf, localFilePath, filename);
                System.out.println("APPEND SUCCESSFULLY");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
public class HDFSReadFile {
    public static void main(String[] args) {
        new ReadFile();
    }
}
class ReadFile {
    public ReadFile() {
        try{
            Configuration conf = new Configuration();
            conf.set("fs.defaultFS", "hdfs://localhost:9000");
            conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
            FileSystem fs = FileSystem.get(conf);
            FSDataInputStream file = fs.open(new Path("/user/hadoop/input/hdfs-site.xml"));
            IOUtils.copyBytes(file, System.out, 1024);
            fs.close();
        }catch(Exception e) {
            e.printStackTrace();
        }
    }
}