Java *** 作Hdfs

Java *** 作Hdfs(分布式文件系统)

文件上传实现

• 创建Maven项目，项目命名Hadoop_hdfs
• 导入pom.xml依赖

<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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0modelVersion>
  <groupId>hadoopgroupId>
  <artifactId>hadoopartifactId>
  <version>0.0.1-SNAPSHOTversion>
  <name>hadoop_learnname>
  
  <dependencies>
  		<!- hadoop配置jar包-->
		<dependency>
		    <groupId>org.apache.hadoopgroupId>
		    <artifactId>hadoop-clientartifactId>
		    <version>3.2.0version>
		dependency>
  dependencies>
  
project>

• 创建HDFS *** 作类

package com.learn_hdfs;


import java.io.FileInputStream;

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 org.apache.hadoop.io.IOUtils;

// *** 作HDFS实现上传hdfs文件
public class HdfsOp {

	//上传文件实现
	public static void main(String[] args) throws Exception{
		//创建一个 *** 作HDFS配置对象
		Configuration config = new Configuration();
		//设置Java代码连接哪个集群的hdfs
		config.set("fs.defaultFS", "hdfs://192.168.72.100:9000");
		//获取 *** 作HDFS的对象
		FileSystem fileSystem = FileSystem.get(config);
		
		//进行上传文件 *** 作
		//获取文件输入流
		FileInputStream fileInputStream =  new FileInputStream("C:\Users\Documents\learn.txt");
		//获取hdfs中的输出流
		FSDataOutputStream fsDataOutputStream =  fileSystem.create(new Path("/learn.txt"));
		
		//第一个参数是一个输入流、第二个参数是输出流，第三个是缓存区大小，第四个参数是设置否关闭流
		IOUtils.copyBytes(fileInputStream, fsDataOutputStream, 1024, true);
		
		
	}
	
}

• 配置好上述功能代码后，通过Linux服务器搭建分布式集群，并修改好集群中的hdfs的访问权限配置。
  通过对hdfs-site.xml进行配置，完成对hdfs的配置

• 启动集群
  
  在本地运行代码后，通过hdfs中的shell命令进行查看。
  

完成上传！

文件下载 *** 作

• 编写下载文件 *** 作类FileHdfsDownLoad类进行实现

package com.learn_hdfs;


import java.io.FileInputStream;
import java.io.FileOutputStream;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

// *** 作HDFS
public class FileHdfsDownLoad{

	//下载
	public static void main(String[] args) throws Exception{
		//创建一个 *** 作HDFS的对象
		Configuration config = new Configuration();
		//设置java代码连接哪个hdfs
		config.set("fs.defaultFS", "hdfs://192.168.72.100:9000");
		//获取 *** 作HDFS的对象
		FileSystem fileSystem = FileSystem.get(config);
		
		
		//下载文件
		downloadFile(fileSystem);
		
		
	}
	//下载hdfs文件
	private static void downloadFile(FileSystem fileSystem) throws Exception{
		//获取hdfs分布式文件系统中的输入流
		FSDataInputStream fsDataInputStream =  fileSystem.open(new Path("/learn.txt"));
		//获取本地文件输出流
		FileOutputStream foFileOutputStream = new FileOutputStream("C:\Users\Documents\learn.txt");
		//下载文件
		IOUtils.copyBytes(fsDataInputStream, foFileOutputStream, 1024, true);
	}
}

• 在启动集群后进行执行程序进行验证是否下载了learn.txt文件
  
  下载成功！

删除文件 *** 作

• 编写删除hdfs文件的函数deleteFile

package com.learn_hdfs;


import java.io.FileInputStream;
import java.io.FileOutputStream;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

// *** 作HDFS
public class HdfsOp {

	//删除
	public static void main(String[] args) throws Exception{
		//创建一个 *** 作HDFS的对象
		Configuration config = new Configuration();
		//设置java代码连接哪个hdfs
		config.set("fs.defaultFS", "hdfs://192.168.72.100:9000");
		//获取 *** 作HDFS的对象
		FileSystem fileSystem = FileSystem.get(config);
		
		
		//删除文件 *** 作
		deleteFile(fileSystem);
		
	}

	
	//删除文件 *** 作
	private static void deleteFile(FileSystem fileSystem) throws Exception{
		boolean result =  fileSystem.delete(new Path("/learn.txt"));
		if(result) {
			System.out.print("删除成功");
		}else {
			System.out.print("删除失败");
		}
	}
	
	
}

• 确保hdfs中存在要删除的文件learn.txt
  
  执行代码
  
• 检查hdfs中是否还存在该文件
  
  不存在该文件，因此删除完毕。

完整代码如下：

import java.io.FileInputStream;
import java.io.FileOutputStream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

// *** 作HDFS
public class HdfsOp {

	//上传、下载、删除
	public static void main(String[] args) throws Exception{
		//创建一个 *** 作HDFS的对象
		Configuration config = new Configuration();
		//设置java代码连接哪个hdfs
		config.set("fs.defaultFS", "hdfs://192.168.72.100:9000");
		//获取 *** 作HDFS的对象
		FileSystem fileSystem = FileSystem.get(config);
		
		//上传hdfs文件
		putFile(fileSystem);
		
		//下载文件
		downloadFile(fileSystem);
		
		//删除文件 *** 作
		deleteFile(fileSystem);
		
	}
	
	//上传文件函数
	private static void putFile(FileSystem fileSystem) throws Exception{
		//进行上传文件 *** 作
		//获取输入流
		FileInputStream fileInputStream =  new FileInputStream("C:\Users\dang\Documents\learn.txt");
		//获取hdfs中的输出流
		FSDataOutputStream fsDataOutputStream =  fileSystem.create(new Path("/learn.txt"));
				
		//第一个参数是一个输入流、第二个参数是输出流，第三个是缓存区大小，第四个参数是是否关闭
		IOUtils.copyBytes(fileInputStream, fsDataOutputStream, 1024, true);
	}
	
	//下载hdfs文件
	private static void downloadFile(FileSystem fileSystem) throws Exception{
		//获取hdfs中的输入流
		FSDataInputStream fsDataInputStream =  fileSystem.open(new Path("/learn.txt"));
		//获取本地文件输出流
		FileOutputStream foFileOutputStream = new FileOutputStream("C:\Users\dang\Documents\learn.txt");
		//下载文件
		IOUtils.copyBytes(fsDataInputStream, foFileOutputStream, 1024, true);
	}
	
	//删除文件 *** 作
	private static void deleteFile(FileSystem fileSystem) throws Exception{
		boolean result =  fileSystem.delete(new Path("/learn.txt"));
		if(result) {
			System.out.print("删除成功");
		}else {
			System.out.print("删除失败");
		}
	}
	
	
}