Hdfs 与 java 连通，实现本地文件与 hdfs 的文件的上传下载

package com.zhiyou100.test;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;

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;

public class HdfsTest1 {
	
	public static final Configuration CONF = new Configuration();

	// 获取分布式文件系统对象
	public static FileSystem hdfs;
	
	static{
		
		try{
			
			hdfs = FileSystem.get(CONF);
		}catch (Exception e){
			
			System.out.println("无法连接 hdfs， 请检查配置");
			
			e.printStackTrace();
		}
	}
	
	/**
	 * 使用字节流从本地读取一个文件，并将读取的文件写到 hdfs 中
	 * @param localPath
	 * @param hdfsPath
	 * @throws Exception 
	 */
	public static void copyFileFromLocation(String localPath, String hdfsPath) throws Exception{
		
		Path path = new Path(hdfsPath);
		
		// 创建本地文件对象
		File file = new File(localPath);
		
		// 如果 path 文件存在，发出提示
		if (hdfs.exists(path)) {
			
			System.out.println("文件" + hdfsPath + "已存在");
		}else{
			
			if (!file.exists()) {
				
				System.out.println("本地文件不存在，请更换路径");
			}else{
				
				// 如果本地文件 file 是一个文件，则复制文件
				if (file.isFile()) {
					
					try {
						// 创建本地文件输入流
						FileInputStream inputStream = new FileInputStream(file);
						
						// 创建 hdfs 中的文件
						FSDataOutputStream outPutStream = hdfs.create(path);
						
						int length = 0;
						
						byte[] bytes = new byte[1024];
						
						// 以字节流读取本地文件
						while((length = inputStream.read(bytes)) != -1){
							
							// 将读入的字节流写入 hdfs 文件中
							outPutStream.write(bytes,0,length);
						}
						
						// 关闭输入流
						inputStream.close();
						
						// 关闭输出流
						outPutStream.close();
						
					} catch (FileNotFoundException e) {

						e.printStackTrace();
					} catch (IOException e) {

						e.printStackTrace();
					}
				}else{
					
					System.out.println("本地目录不是一个文件，请更换目录");
				}
			}
			
			
		}
	}
	
	/**
	 * 从 hdfs 中读取一个文件，并将读取的文件写到本地
	 * @param localPath
	 * @param hdfsPath
	 * @throws IOException 
	 */
	public static void copyFileToLocation(String localPath, String hdfsPath) throws IOException{
		
		Path path = new Path(hdfsPath);
		
		File file = new File(localPath);
		
		if (!hdfs.exists(path)) {
			
			System.out.println("hdfs 中文件不存在，请更换目录");
		}else{
			
			if(file.exists() || file.isDirectory()){
				
				System.out.println("本地文件已存在或该文件是一个文件夹，请更换目录");
			}else{
				
				try {
					FSDataInputStream inputStream = hdfs.open(path);
					
					FileOutputStream outPutStream = new FileOutputStream(file);
					
					int length = 0;
					
					byte[] bytes = new byte[1024];
					
					while((length = inputStream.read(bytes)) != -1) {
						
						outPutStream.write(bytes, 0, length);
					}
					
					outPutStream.close();
					
					inputStream.close();
					
				} catch (IOException e) {

					e.printStackTrace();
				}
			}
			}
			
			
		
		
	}
	
	public static void main(String[] args) {
		
		try {
			copyFileFromLocation("D:\\JavaTest\\JavaTest3\\三国.txt", "/test/三国");
			
		} catch (Exception e) {

			e.printStackTrace();
		}
		
		try {

			copyFileToLocation("D:\\JavaTest\\大数据预习.txt", "/dirFromJava/大数据预习.txt");

		} catch (IOException e) {

			e.printStackTrace();
		}
	}
}