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  //从本地复制文件到HDFShadoop   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）   //将本地文件上传到hdfshadoop   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 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();        }    }}