掌握HDFS的Shell访问和Java API访问

         HDFS设计的主要目的是对海量数据进行处理，也就是说在其上能够存储很大量文件（可以存储TB级的文件）。HDFS将这些文件分割之后，存储在不同的DataNode上， HDFS 提供了两种数据访问接口：Shell接口和Java API 接口，对HDFS里面的文件进行操作，具体每个Block放在哪台DataNode上面，对于开发者来说是透明的。

Shell 接口

        HDFS 处理文件的命令和 Linux 命令基本相同，这里区分大小写。下面介绍 HDFS 操作分布式文件系统的命令。

        HDFS基本命令:

hadoop fs -cmd

cmd:具体的操作，基本上与UNIX的命令行相同

args:参数

        HDFS 资源URI格式：

scheme://authority/path

scheme：协议名，file或hdfs

authority：namenode主机名

path：路径 

示例：hdfs://cloudxxx:9000/middle/test.txt

假设已经在core-site.xml里配置了fs.default.name=hdfs://cloudxxx:9000，则仅使用/middle/test.txt即可。

shell 操作单个 HDFS 集群

        下面列举出几个常用场景下的命令。

        1、创建文件夹

        HDFS 上的文件目录结构类似 Linux，根目录使用 "/" 表示。下面的命令将在 /middle 目录下建立目录 weibo

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fs -mkdir  /middle/weibo

        2、上传文件 weibo.txt 到 weibo 目录下。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fs -put weibo.txt /middle/weibo/

        还可以使用 -copyFromLocal 参数。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fs -copyFromLocal weibo.txt /middle/weibo/

        3、查看 weibo.txt 文件内容。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fs -text /middle/weibo/weibo.txt

        还可以用 -cat、-tail 参数查看文件的内容。但是对于压缩的结果文件只能用 -text 参数来查看，否则是乱码。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fs -cat /middle/weibo/weibo.txt

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fs -tail /middle/weibo/weibo.txt

        4、把 weibo.txt 文件复制到本地。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fs -get /middle/weibo/weibo.txt

        还可以用 -copyToLocal 参数。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fs -copyToLocal /middle/weibo/weibo.txt

        5、删除 weibo.txt 文件。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fs -rm /middle/weibo/weibo.txt

        删除/middle/weibo文件夹。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fs -rmr /middle/weibo

        6、显示 /middle 目录下的文件。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fs -ls /middle

shell 操作多个 HDFS 集群

        上面我们介绍的是单线程访问的 HDFS 访问模型，但是 多个Hadoop 集群需要复制数据该怎么办呢？幸运的是，Hadoop 有一个有用的 distcp 分布式复制程序，该程序是由 MapReduce 作业来实现的，它是通过集群中并行运行的 map 来完成集群之间大量数据的复制。 下面我们将介绍 distcp 在不同场景下该如何使用。

        两个集群运行相同版本的 Hadoop。

        两个 HDFS 集群之间传输数据，默认情况下 distcp 会跳过目标路径下已经存在的文件。

[hadoop@cloudxxxx hadoop-2.2.0-x64]$ hadoop distcp hdfs://cloudxxx:9000/weather hdfs://cloudxxxx:9000/middle

        这条指令把第一个集群/weather 目录及其内容复制到第二个集群的/middle 目录下，所以第二个集群最后的目录结构为/middle/weather。如果/middle 不存在，则新建一个。 也可以指定多个源路径，并把所有路径都复制到目标路径下。这里的源路径必须是绝对路径。

        两个 HDFS 集群之间传输数据，覆盖现有的文件使用overwrite。

[hadoop@cloudxxxx hadoop-2.2.0-x64]$ hadoop distcp -overwrite hdfs://cloudxxx:9000/weather hdfs://cloudxxxx2:9000/middle

        两个 HDFS 集群之间传输数据，更新有改动过的文件使用update。

[hadoop@cloudxxxx hadoop-2.2.0-x64]$ hadoop distcp -update hdfs://cloudxxx:9000/weather hdfs://cloudxxxx:9000/middle

        两个集群运行不同版本的 Hadoop。

        不同版本 Hadoop 集群的 RPC 是不兼容的，使用distcp 复制数据并使用 hdfs 协议，会导致复制作业失败。想要弥补这种情况，可以使用基于只读 HTTP 协议的 HFTP 文件系统并从源文件系统中读取数据。 这个作业必须运行在目标集群上，进而实现 hdfs RPC 版本的兼容。

        还以两个 HDFS 集群之间传输数据为例。

[hadoop@cloudxxxx hadoop-2.2.0-x64]$ hadoop distcp hftp://cloudxxx:9000/weather hdfs://cloudxxxx:9000/middle

        注意，这里需要在 URI 源中指定 namenode 的 Web 端口。这是由 dfs.http.address 属性决定的，其默认值为 50070。

        如果使用新出的 webhdfs 协议（替代 hftp）后，对源集群和目标集群均可以使用 HTTP 协议进行通信，且不会造成任何不兼容的问题。

[hadoop@cloudxxxx hadoop-2.2.0-x64]$ hadoop distcp webhdfs://cloudxxx:9000/weather webhdfs://cloudxxxx:9000/middle

Hadoop 管理员其他常见shell操作

        掌握了 shell 如何访问 HDFS，作为 Hadoop 管理员，还需要掌握如下常见命令：

        1、查看正在运行的 Job。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop job -list

        2、关闭正在运行的 Job。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop job -kill job_1432108212572_0001

        3、检查 HDFS 块状态，查看是否损坏。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fsck /

        4、检查 HDFS 块状态，并删除损坏的块。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop fsck / -delete

        5、检查 HDFS 状态，包括 DataNode 信息。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop dfsadmin -report

        6、Hadoop 进入安全模式。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop dfsadmin -safemode enter

        7、Hadoop 离开安全模式。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop dfsadmin -safemode leave

        8、平衡集群中的文件

[hadoop@cloudxxx hadoop-2.2.0-x64]$ sbin/start-balancer.sh

        如果无法直接使用 hadoop fs -cmd 命令，请查看这里。想了解更多的 HDFS 的Shell 命令，请查看命令大全

Java API接口

        HDFS提供了Java API接口对HDFS进行操作。如果以下程序在 Hadoop 集群上运行， Path 中的路径可以写为相对路径，比如 "/middle/weibo"； 如果以下程序在本地 Eclipse 上面测试，Path 中的路径需要写为绝对路径，比如"hdfs://cloudxxx:9000/middle/weibo"。

        1、获取 HDFS 文件系统

//获取文件系统

public static FileSystem getFileSystem() throws IOException {

//读取配置文件

Configuration conf = new Configuration();

//返回默认文件系统  如果在 Hadoop集群下运行，使用此种方法可直接获取默认文件系统

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

//指定的文件系统地址

URI uri = new URI("hdfs://cloudxxx:9000");

//返回指定的文件系统    如果在本地测试，需要使用此种方法获取文件系统

FileSystem fs = FileSystem.get(uri,conf); 

return fs;

}

        如果放到 hadoop 集群上面运行，获取文件系统可以直接使用 FileSystem.get(conf)。

        2、创建文件目录

//创建文件目录

public static void mkdir() throws Exception {

//获取文件系统

FileSystem fs = getFileSystem();

//创建文件目录

fs.mkdirs(new Path("hdfs://cloudxxx:9000/middle/weibo"));

//释放资源

fs.close();

}

        3、删除文件或文件目录

//删除文件或者文件目录

public static void rmdir() throws Exception {

//返回FileSystem对象

FileSystem fs = getFileSystem();

//删除文件或者文件目录  delete(Path f) 此方法已经弃用

fs.delete(new Path("hdfs://cloudxxx:9000/middle/weibo"),true);

//释放资源

fs.close();

}

        4、获取目录下的所有文件

//获取目录下的所有文件

public static void ListAllFile() throws IOException{

//返回FileSystem对象

FileSystem fs = getFileSystem();

//列出目录内容

FileStatus[] status = fs.listStatus(new Path("hdfs://cloudxxx:9000/middle/weibo/"));

//获取目录下的所有文件路径

Path[] listedPaths = FileUtil.stat2Paths(status);

//循环读取每个文件

for(Path p : listedPaths){

System.out.println(p);

}

//释放资源

fs.close();

}

        5、文件上传至 HDFS

//文件上传至 HDFS

public static void copyToHDFS() throws IOException{

//返回FileSystem对象

FileSystem fs = getFileSystem();

//源文件路径是Linux下的路径，如果在 windows 下测试，需要改写为Windows下的路径，比如D://hadoop/djt/weibo.txt

Path srcPath = new Path("/home/hadoop/djt/weibo.txt");

// 目的路径

Path dstPath = new Path("hdfs://cloudxxx:9000/middle/weibo");

//实现文件上传

fs.copyFromLocalFile(srcPath, dstPath);

//释放资源

fs.close();

}

        6、从 HDFS 下载文件

//从 HDFS 下载文件

public static void getFile() throws IOException{

//返回FileSystem对象

FileSystem fs = getFileSystem();

//源文件路径

Path srcPath = new Path("hdfs://cloudxxx:9000/middle/weibo/weibo.txt");

//目的路径是Linux下的路径，如果在 windows 下测试，需要改写为Windows下的路径，比如D://hadoop/djt/

Path dstPath = new Path("/home/hadoop/djt/");

//下载hdfs上的文件

fs.copyToLocalFile(srcPath, dstPath);

//释放资源

fs.close();

}

        7、获取 HDFS 集群节点信息

//获取 HDFS 集群节点信息

public static void getHDFSNodes() throws IOException{

//返回FileSystem对象

FileSystem fs = getFileSystem();

//获取分布式文件系统

DistributedFileSystem hdfs = (DistributedFileSystem)fs;

//获取所有节点

DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats();

//循环打印所有节点

for(int i=0;i< dataNodeStats.length;i++){

System.out.println("DataNode_"+i+"_Name:"+dataNodeStats[i].getHostName());

}

}

        8、查找某个文件在 HDFS 集群的位置

//查找某个文件在 HDFS 集群的位置

public static void getFileLocal() throws IOException{

//返回FileSystem对象

FileSystem fs = getFileSystem();

//文件路径

Path path = new Path("hdfs://cloudxxx:9000/middle/weibo/weibo.txt");

//获取文件目录

FileStatus filestatus = fs.getFileStatus(path);

//获取文件块位置列表

BlockLocation[] blkLocations = fs.getFileBlockLocations(filestatus, 0, filestatus.getLen());

//循环输出块信息

for(int i=0;i< blkLocations.length;i++){

String[] hosts = blkLocations[i].getHosts();

System.out.println("block_"+i+"_location:"+hosts[0]);

}

}

        上述 Java API对 HDFS 的操作方法，可以逐个放在 Test 类中进行测试。

package com.dajiangtai.hadoop.middle;

import java.io.IOException;

import java.net.URI;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.BlockLocation;

import org.apache.hadoop.fs.FileStatus;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.FileUtil;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.hdfs.DistributedFileSystem;

import org.apache.hadoop.hdfs.protocol.DatanodeInfo;

public class Test {

public static void main(String[] args) throws Exception {

// TODO Auto-generated method stub

//mkdir();

//rmdir();

//ListAllFile();

//copyToHDFS();

//getFile();

//getHDFSNodes();

//getFileLocal();

}

}

运行程序

        以上 Java API 操作 HDFS 的各种方法，在本地测试完成后，可以根据自己实际情况，稍加修改Path路径就可以直接放到 Hadoop 环境下运行。大致需要以下几步：

        第一步：我们使用 Eclipse 将 Test.java 类打包为 test.jar。因为这里不涉及第三方 jar 包，而且 Hadoop 集群上面已经存在 hadoop 所需要的jar包，所以我们只需要打包 Test.java 即可。

test.jar

        第二步：在root用户下，通过 xshell 客户端，将 test.jar 上传至 hadoop服务器本地目录/home/hadoop/djt/的下面。

[root@cloudxxx djt]$ ls

test.jar

        第三步：需要在 root 用户下，将我们上传的 test.jar 文件权限赋予 hadoop 用户组。

[root@cloudxxx ~]# chown -R hadoop:hadoop /home/hadoop/djt/test.jar

        第四步：我们需要切换到 hadoop 的安装目录（/usr/java/hadoop-2.2.0-x64）下来运行 test.jar 文件，否则无法找到需要的 jar包来执行程序。

[hadoop@cloudxxx hadoop-2.2.0-x64]$ hadoop jar /home/hadoop/djt/test.jar com.dajiangtai.hadoop.middle.Test