Java操作Hadoop-HDFS详解_java 文件上传hdfs不需要配置hadoop

Tip:本节内容均是在我之前搭建的完全分布式上进行！！

Java客户端操作HDFS

	服务端：  启动NN,DN
	客户端：  使用shell客户端   hadoop fs
			  使用java客户端    
			  使用python客户端
1
2
3
4

我在windows也要配置hadoop，如下图：


然后创建一个maven项目，并添加依赖

<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/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.bjsxt</groupId>
  <artifactId>hadoop</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <name>hadoop</name>
<dependencies>
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>RELEASE</version>
		</dependency>
		<dependency>
			<groupId>org.apache.logging.log4j</groupId>
			<artifactId>log4j-core</artifactId>
			<version>2.8.2</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-common</artifactId>
			<version>2.7.2</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-client</artifactId>
			<version>2.7.2</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-hdfs</artifactId>
			<version>2.7.2</version>
		</dependency> 
<dependency>
            <groupId>jdk.tools</groupId>
            <artifactId>jdk.tools</artifactId>
            <version>1.8</version>
            <scope>system</scope>
            <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
    </dependency>
</dependencies>
</project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

使用Java客户端创建目录

1. FileSystem: 文件系统的抽象基类
 * 			FileSystem的实现取决于fs.defaultFS的配置！
 * 				
 * 		有两种实现！
 *     LocalFileSystem： 本地文件系统   fs.defaultFS=file:///
 *     DistributedFileSystem： 分布式文件系统  fs.defaultFS=hdfs://xxx:9000
 *     
 *     声明用户身份：
 *     	 FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), conf, "atguigu");
 *     
 *  2. Configuration : 功能是读取配置文件中的参数
 *  		
		Configuration在读取配置文件的参数时，根据文件名，从类路径按照顺序读取配置文件！
				先读取 xxx-default.xml，再读取xxx-site.xml
				
		Configuration类一加载，就会默认读取8个配置文件！
		将8个配置文件中所有属性，读取到一个Map集合中！
		
		也提供了set(name,value)，来手动设置用户自定义的参数！
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

@Test
	//hadoop fs -mkdir /xxx
	//创建一个客户端对象，mkdir是个方法，然后还要传个路径
	public void MkDir() throws IOException{
		Configuration conf=new Configuration();
		FileSystem fSystem=FileSystem.get(conf);
		System.out.println(fSystem.getClass().getName());
	}
1
2
3
4
5
6
7
8

如上图，我们不指定配置文件时系统默认加载默认配置文件，默认配置文件里设置的是本地文件系统，对应的就是LocalFileSystem类。

下面我们指定配置文件core-site.xml。这时就是分布式文件系统了，我这里的hadoop1是搭在linux上的。
下面我们来创建目录。

public void MkDir() throws IOException{
		Configuration conf=new Configuration();
		FileSystem fSystem=FileSystem.get(conf);
		fSystem.mkdirs(new Path("/eclipse"));
		System.out.println(fSystem.getClass().getName());
	}
1
2
3
4
5
6

然后我们会发现凉了，因为我windows的用户是联想，而分布式文件系统的拥有者是linux里的ygp。
那么我们就换一个重载的方法，如下代码：

public void MkDir() throws IOException, InterruptedException, URISyntaxException{
		Configuration conf=new Configuration();
		//FileSystem fSystem=FileSystem.get(conf);
		FileSystem fSystem=FileSystem.get(new URI("hdfs://hadoop1:9000"), conf, "ygp");
		fSystem.mkdirs(new Path("/eclipse"));
		System.out.println(fSystem.getClass().getName());
	}
1
2
3
4
5
6
7

创建成功
[注]如果在java代码里指定了uri，那么就没必要在配置文件里配置了。

上传和下载

我把下面这篇文章上传到分布式文件系统的根目录下

// 上传文件： hadoop fs -put 本地文件  hdfs
public void Upload() throws IOException, InterruptedException, URISyntaxException{
	Configuration conf=new Configuration();
	//FileSystem fSystem=FileSystem.get(conf);
	FileSystem fSystem=FileSystem.get(new URI("hdfs://hadoop1:9000"), conf, "ygp");
     fSystem.copyFromLocalFile(false, true, new Path("C:/Users/Lenovo/Desktop/乔老师/05534745.pdf"), new Path("/"));//这个方法的第一个参数是上传完成后是否删除源文件，第二个参数是否覆盖分布式文件系统的同名文件
	fSystem.close();
	
}
1
2
3
4
5
6
7
8
9

成功
同理，来看一下下载

//下载文件：  hadoop fs -get hdfs  本地路径
public void Download() throws IOException, InterruptedException, URISyntaxException{
	Configuration conf=new Configuration();
	//FileSystem fSystem=FileSystem.get(conf);
	FileSystem fSystem=FileSystem.get(new URI("hdfs://hadoop1:9000"), conf, "ygp");
     fSystem.copyToLocalFile(false, new Path("/05534745.pdf"), new Path("C:/Users/Lenovo/Desktop/stanford"), false);//第一个参数表示是否删除源文件，第四个参数如果为false，会生成一个校验文件
	fSystem.close();
}
1
2
3
4
5
6
7
8

删除和重命名

删除某文件或文件夹

// 删除文件：  hadoop fs -rm -r -f  路径
public void Delete() throws IOException, InterruptedException, URISyntaxException{
	Configuration conf=new Configuration();
	//FileSystem fSystem=FileSystem.get(conf);
	FileSystem fSystem=FileSystem.get(new URI("hdfs://hadoop1:9000"), conf, "ygp");
	fSystem.delete(new Path("/eclipse"),true);//第二个参数是删除文件夹是是否递归删除
	fSystem.close();
	
}
1
2
3
4
5
6
7
8
9

成功！！
对文件或文件夹重命名

// 重命名：  hadoop fs -mv  源文件   目标文件
public void Rename() throws IOException, InterruptedException, URISyntaxException{
	Configuration conf=new Configuration();
	//FileSystem fSystem=FileSystem.get(conf);
	FileSystem fSystem=FileSystem.get(new URI("hdfs://hadoop1:9000"), conf, "ygp");
	fSystem.rename(new Path("/05534745.pdf"), new Path("/A Novel Predictor for Moving Objects.pdf"));
	fSystem.close();
	
}
1
2
3
4
5
6
7
8
9

成功！！

判断路径是否存在

public void IfPathExsits() throws IOException, InterruptedException, URISyntaxException{
	Configuration conf=new Configuration();
	//FileSystem fSystem=FileSystem.get(conf);
	FileSystem fSystem=FileSystem.get(new URI("hdfs://hadoop1:9000"), conf, "ygp");
	System.out.println(fSystem.exists(new Path("/A Novel Predictor for Moving Objects.pdf")));
	System.out.println(fSystem.exists(new Path("/1234.pdf")));
	fSystem.close();
	
}
1
2
3
4
5
6
7
8
9

判断是文件还是目录

public void ifDirOrFile() throws IOException, InterruptedException, URISyntaxException{
	Configuration conf=new Configuration();
	//FileSystem fSystem=FileSystem.get(conf);
	FileSystem fSystem=FileSystem.get(new URI("hdfs://hadoop1:9000"), conf, "ygp");
	Path path=new Path("/A Novel Predictor for Moving Objects.pdf");
	System.out.println(fSystem.isDirectory(path));
	System.out.println(fSystem.isFile(path));
	fSystem.close();
	
}
1
2
3
4
5
6
7
8
9
10

但是官方不推荐用上述的做法，而推荐用下述的做法，使用FileStatus对象来判断。

public void ifDirOrFile() throws IOException, InterruptedException, URISyntaxException{
	Configuration conf=new Configuration();
	//FileSystem fSystem=FileSystem.get(conf);
	FileSystem fSystem=FileSystem.get(new URI("hdfs://hadoop1:9000"), conf, "ygp");
	Path path=new Path("/A Novel Predictor for Moving Objects.pdf");
	FileStatus fileStatus = fSystem.getFileStatus(path);
	System.out.println("是否是目录："+fileStatus.isDirectory());
	System.out.println("是否是文件："+fileStatus.isFile());
	fSystem.close();
	
}
1
2
3
4
5
6
7
8
9
10
11

官方还提供了一个超级强大的方法，为了说明此方法，我需要建个目录，还是就叫eclipse吧，然后在里面分别建立三个文件，file1，file2，file3，如下图：

public void ifDirOrFile() throws IOException, InterruptedException, URISyntaxException{
	Configuration conf=new Configuration();
	//FileSystem fSystem=FileSystem.get(conf);
	FileSystem fSystem=FileSystem.get(new URI("hdfs://hadoop1:9000"), conf, "ygp");
	Path path=new Path("/eclipse");
	FileStatus[] listStatus = fSystem.listStatus(path);
	
	for (FileStatus fileStatus : listStatus) {
		
		//获取文件名 Path是完整的路径 协议+文件名
		Path filePath = fileStatus.getPath();
		System.out.println(filePath.getName()+"是否是目录："+fileStatus.isDirectory());
		System.out.println(filePath.getName()+"是否是文件："+fileStatus.isFile());
	}
	fSystem.close();
	
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

获取块的位置信息

现在我把一个大于670M的文件传入到HDFS中，看看会有什么后果


如上图所示，我们发现每一个块默认128M，然后670M会分成5个块

offset和length
			offset是偏移量： 指块在文件中的起始位置
			length是长度，指块大小
			
			PPT.zip 670M
								length    offset
			blk1:   0-128M      128M		0
			blk2:    128M-256M  128M        128M
			...
			blk5:    ...-670M   ...        ...
1
2
3
4
5
6
7
8
9
10

public void testGetBlockInfomation() throws Exception {
		
		Path path = new Path("/PPT.zip");
		
		RemoteIterator<LocatedFileStatus> status = fs.listLocatedStatus(path);
		
		while(status.hasNext()) {
			
			LocatedFileStatus locatedFileStatus = status.next();
			
			System.out.println("属主:"+locatedFileStatus.getOwner());
			System.out.println("属组:"+locatedFileStatus.getGroup());
			
			//---------------块的位置信息--------------------
			BlockLocation[] blockLocations = locatedFileStatus.getBlockLocations();
			
			for (BlockLocation blockLocation : blockLocations) {
				
				System.out.println(blockLocation);
				
				System.out.println("------------------------");
				
			}
			
		}
		
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

[注]如果你传的文件再大点，导致你的CPU消耗过大，分块这个操作可能会出问题，因为你的datanode也许会自动关闭。