１. HDFS的java操作
hdfs在生产应用中主要是客户端的开发，其核心步骤是从hdfs提供的api中构造一个HDFS的访问客户端对象，然后通过该客户端对象操作（增删改查）HDFS上的文件
１.1 搭建开发环境
１.1.1创建Maven工程
快速创建一个Maven工程和目录结构的方式是执行下面的命令：
mvn archetype:generate -DgroupId=com.toto.hadoop -DartifactId=hadoop -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false |
| 接着执行下面的命令： mvn -Pall eclipse:eclipse mvn clean mvn compile -Dmaven.test.skip=true mvn install -Dmaven.test.skip=true mvn package -Dmaven.test.skip=true |
最后将工程导入到Eclipse中： |
1、配置pom文件，引入下面的jar之后，其它相关的maven依赖的jar包，它会自动引入：
| <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/maven-v4_0_0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.toto.hadoop</groupId> <artifactId>hadoop</artifactId> <packaging>jar</packaging> <version>1.0-SNAPSHOT</version> <name>hadoop</name> <url>http://maven.apache.org</url> <dependencies> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.5.1</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.5.1</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.5.1</version> </dependency>
<dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.12</version> <scope>test</scope> </dependency> </dependencies> </project> |
１.1.2创建一个非Maven工程
注：如需手动引入jar包，hdfs的jar包----hadoop的安装目录的share下，他们分别是：
| hadoop-2.8.0\share\hadoop\common\hadoop-common-2.8.0.jar hadoop-2.8.0\share\hadoop\common\lib hadoop-2.8.0\share\hadoop\hdfs\hadoop-hdfs-2.8.0.jar hadoop-2.8.0\share\hadoop\hdfs\lib |
如果是创建非Maven工程，步骤如下：
第一步：创建一个User Library
第二步：添加library
第三步：添加common中的所有的jar
第四步：添加hadoop-hdfs-2.8.0.jar
第五步：添加hdfs依赖的jar
最后到工程里面查看引包情况
接着上面的配置，可以进行hdfs客户端调用的代码的编写了。
2、window下开发的说明
建议在linux下进行hadoop应用的开发，不会存在兼容性问题。如在window上做客户端应用开发，需要设置以下环境：
A、用老师给的windows平台下编译的hadoop安装包解压一份到windows的任意一个目录下
B、在window系统中配置HADOOP_HOME指向你解压的安装包目录
C、在windows系统的path变量中加入HADOOP_HOME的bin目录
１.2 获取api中的客户端对象
在java中操作hdfs，首先要获得一个客户端实例
| Configuration conf = new Configuration() FileSystem fs = FileSystem.get(conf) |
而我们的操作目标是HDFS，所以获取到的fs对象应该是DistributedFileSystem的实例；
get方法是从何处判断具体实例化那种客户端类呢？
——从conf中的一个参数 fs.defaultFS的配置值判断；
如果我们的代码中没有指定fs.defaultFS，并且工程classpath下也没有给定相应的配置，conf中的默认值就来自于hadoop的jar包中的core-default.xml，默认值为： file:///，则获取的将不是一个DistributedFileSystem的实例，而是一个本地文件系统的客户端对象
１.3 DistributedFileSystem实例对象所具备的方法
１.4 HDFS客户端操作数据代码示例：
１.4.1 文件的增删改查
| package com.toto.hadooptest;
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.LocatedFileStatus; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.RemoteIterator; import org.junit.Before; import org.junit.Test;
public class HdfsClient {
FileSystem fs = null;
@Before public void init() throws Exception { // 构造一个配置参数对象，设置一个参数：我们要访问的hdfs的URI // 从而FileSystem.get()方法就知道应该是去构造一个访问hdfs文件系统的客户端，以及hdfs的访问地址 // new Configuration();的时候，它就会去加载jar包中的hdfs-default.xml // 然后再加载classpath下的hdfs-site.xml Configuration conf = new Configuration(); conf.set("df.defaultFS", "hdfs://hadoop:9000"); /** * 参数优先级：1、客户端代码中设置的值 2、classpath下的用户自定义配置文件 3、然后是服务器的默认配置 */ conf.set("dfs.replication", "3");
// 获取一个hdfs的访问客户端，根据参数，这个实例应该是DistributedFileSystem的实例 // fs = FileSystem.get(conf);
//如果这样去获取，那conf里面就可以不要配"fs.defaultFS"参数，而且，这个客户端的身份识别已经是hadoop用户 //注意，最后的toto表示的是hadoop集群安装的Linux用户 fs = FileSystem.get(new URI("hdfs://hadoop:9000"),conf,"toto"); }
/** * \brief hdfs上传文件 * * 执行完成之后，进入Linux，执行一下命令，发现会出现以下列表 * [toto@hadoop software]$ hadoop fs -ls /toto2 Found 1 items -rw-r--r-- 3 toto supergroup 2286088 2017-05-30 13:25 /toto2/ik-analyzer-solr5-master.zip */ @Test public void testAddFileToHdfs() { try { //要上传的所在的本地路径 Path src = new Path("d:/ik-analyzer-solr5-master.zip"); //要上传的hdfs的目标路径 Path dst = new Path("/toto2"); fs.copyFromLocalFile(src, dst); fs.close(); } catch (Exception e) { e.printStackTrace(); } }
/** * \brief 从hdfs中复制文件到本地文件系统 * @attention执行完成之后，进入E:/learnTempFolder这个文件夹，发现在改文件夹下已经有了自己拷贝下来的文件了 * @author toto * @date 2017年5月30日 * @note begin modify by 涂作权 2017年5月30日 原始创建 */ @Test public void testDownLoadFileToLocal() { try { fs.copyToLocalFile(new Path("/toto2/findbugs-1.3.9"), new Path("E:/learnTempFolder")); fs.close(); } catch (Exception e) { e.printStackTrace(); } }
/** * 创建目录，删除目录，重命名 * 执行之前hadoop的文件系统的目录结构如下： * [toto@hadoop sbin]$ hadoop fs -ls / Found 7 items drwxr-xr-x - toto supergroup 0 2017-05-30 14:06 /aaa drwxr-xr-x - toto supergroup 0 2017-05-29 14:01 /findbugs-1.3.9 drwxr-xr-x - toto supergroup 0 2017-05-29 03:23 /hive drwx------ - toto supergroup 0 2017-05-29 14:47 /tmp drwxr-xr-x - toto supergroup 0 2017-05-29 23:59 /toto drwxr-xr-x - toto supergroup 0 2017-05-30 13:25 /toto2 drwxr-xr-x - toto supergroup 0 2017-05-30 00:18 /user [toto@hadoop sbin]$
执行之后的效果： [toto@hadoop sbin]$ hadoop fs -ls / Found 7 items drwxr-xr-x - toto supergroup 0 2017-05-30 14:11 /a2 drwxr-xr-x - toto supergroup 0 2017-05-29 14:01 /findbugs-1.3.9 drwxr-xr-x - toto supergroup 0 2017-05-29 03:23 /hive drwx------ - toto supergroup 0 2017-05-29 14:47 /tmp drwxr-xr-x - toto supergroup 0 2017-05-29 23:59 /toto drwxr-xr-x - toto supergroup 0 2017-05-30 13:25 /toto2 drwxr-xr-x - toto supergroup 0 2017-05-30 00:18 /user
总结： 1、发现/aaa文件夹被删除了 2、发现出现了/a2,但是最开始没有，说明是创建的a1,然后又被改成的a2 */ @Test public void testMkdirDeleteAndRename() { try { //创建目录 fs.mkdirs(new Path("/a1/b1/c1"));
//删除文件夹，如果是非空文件夹，参数2必须给值true fs.delete(new Path("/aaa"),true);
//重命名文件或文件夹 fs.rename(new Path("/a1"), new Path("/a2")); } catch(Exception e) { e.printStackTrace(); } }
/** * \brief 查看目录信息，只显示文件 */ @Test public void testListFiles() { try { //思考：为什么返回迭代器，而不是List之类的容器，这里的只有调用hasNext()的时候，才会返回实际的数据信息 RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
while(listFiles.hasNext()) { LocatedFileStatus fileStatus = listFiles.next(); System.out.println(fileStatus.getPath().getName()); System.out.println(fileStatus.getBlockSize()); System.out.println(fileStatus.getPermission()); System.out.println(fileStatus.getLen()); BlockLocation[] blockLocations = fileStatus.getBlockLocations(); for(BlockLocation b1 : blockLocations) { System.out.println("block-length:" + b1.getLength() + "--" + "block-offset:" + b1.getOffset()); String[] hosts = b1.getHosts(); for(String host : hosts) { System.out.println(host); } } System.out.println("-----------------------------------"); } } catch (Exception e) { e.printStackTrace(); } }
/** * 查看文件及文件夹信息 * * 显示的内容如下： * d-- a2 * d-- findbugs-1.3.9 * d-- hive * d-- tmp * d-- toto * d-- toto2 * d-- user */ @Test public void testListAll() { try { FileStatus[] listStatus = fs.listStatus(new Path("/"));
String flag = "d-- "; for(FileStatus fstatus : listStatus) { if (fstatus.isFile()) flag = "f-- "; System.out.println(flag + fstatus.getPath().getName()); } } catch (Exception e) { e.printStackTrace(); } } } |
１.4.2 通过流的方式访问hdfs
| package com.toto.hadooptest;
import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.net.URI;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.BlockLocation; import org.apache.hadoop.fs.FSDataInputStream; 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.junit.Before; import org.junit.Test;
/** * 相对那些封装好的方法而言的更底层一些的操作方式 * 上层那些mapreduce,spark等运算框架，去hdfs中获取数据的时候，就是调用这种底层的api */ public class StreamAccess { FileSystem fs = null;
@Before public void init() throws Exception {
Configuration conf = new Configuration(); fs = FileSystem.get(new URI("hdfs://hadoop:9000"), conf, "toto");
}
/** * 通过流的方式上传文件到hdfs * @throws Exception */ @Test public void testUpload() throws Exception { FSDataOutputStream outputStream = fs.create(new Path("/README.md"), true); FileInputStream inputStream = new FileInputStream("E:/learnTempFolder/dubbo-master/README.md");
//注意这里的IOUtils是org.apache.hadoop.io.IOUtils中的包，否则创建不成功 IOUtils.copyBytes(inputStream, outputStream, 4096); }
/** * 从hdfs上下载资源文件到本地 */ @Test public void testDownLoadFileToLocal() { try{ //先获取一个文件的输入流---针对hdfs上的 FSDataInputStream in = fs.open(new Path("/hive/apache-hive-2.0.0-bin.tar.gz"));
//再构造一个文件的输出流---针对本地的 FileOutputStream out = new FileOutputStream(new File("E:/apache-hive-2.0.0-bin.tar.gz"));
//再将输入流中数据传输到输出流 IOUtils.copyBytes(in, out, 4096); } catch (Exception e) { e.printStackTrace(); } }
/** * hdfs支持随机定位进行文件读取，而且可以方便地读取指定长度 * 用于上层分布式运算框架并发处理数据 */ @Test public void testRandomAccess() { try { //先获取一个文件的输入流---针对hdfs上的 FSDataInputStream in = fs.open(new Path("/pom.xml"));
//可以将流的起始偏移量进行自定义 in.seek(200);
//在构造一个文件的输出流----针对本地的 FileOutputStream out = new FileOutputStream(new File("E:/pom.xml"));
//第三个参数的意思是获取长度为3000的文本内容 IOUtils.copyBytes(in, out,3000L, true); } catch (Exception e) { e.printStackTrace(); } }
/** * 显示hdfs上文件的内容 */ @Test public void testCat() { try { FSDataInputStream in = fs.open(new Path("/pom.xml"));
IOUtils.copyBytes(in, System.out, 1024); } catch (Exception e) { e.printStackTrace(); } }
} |
１.4.3 场景编程
在mapreduce 、spark等运算框架中，有一个核心思想就是将运算移往数据，或者说，就是要在并发计算中尽可能让运算本地化，这就需要获取数据所在位置的信息并进行相应范围读取以下模拟实现：获取一个文件的所有block位置信息，然后读取指定block中的内容
| @Test public void testCat2() { try { FSDataInputStream in = fs.open(new Path("/toto2/findbugs-1.3.9")); //拿到文件信息 FileStatus[] listStatus = fs.listStatus(new Path("/toto2/findbugs-1.3.9")); //获取这个文件的所有的block的信息 BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(listStatus[0], 0, listStatus[0].getLen()); //第一个block的长度 long length = fileBlockLocations[0].getLength(); //第一个block的起始偏移量 long offset = fileBlockLocations[0].getOffset();
System.out.println(length); System.out.println(offset);
//获取第一个block写入输出流 IOUtils.copyBytes(in, System.out, (int)length); byte[] b = new byte[4096];
@SuppressWarnings("resource") FileOutputStream os = new FileOutputStream(new File("e:/block")); while(in.read(offset,b,0,4096) != -1) { os.write(b); offset += 4096; if (offset >= length) return; }
os.flush(); os.close(); in.close(); } catch (Exception e) { e.printStackTrace(); } } |
