Hadoop 之 HDFS操作

HDFS的设计

HDFS 以流式数据访问模式来存储超大文件，运行于商用硬件集群上。HDFS是分布式文件系统

HDFS的数据块

HDFS的数据块一般的大小是128MB
PS： 寻址速度10ms 硬盘读取速度100MB/s 一般寻址速度是数据IO的百分之一
我们以拷贝一个大于128M的文件为例子
步骤一： 首先找到大于128MB的文件，将文件拷贝到hdfs下，hdfs只是一个逻辑上的文件系统
hdfs dfs -copyFromLocal /Users/zzjmay/Downloads/pycharm-professional-2016.3.1.dmg /zzjmay/hadoop/
本地文件拷贝 copyFromLocal，可以通过WebUI看到，一个268MB的文件被切割成了三个数据块，其中需要说明的是BlockId是其在数据节点中存储的ID

HDFS的Namenode和datanode

NameNode: 名称节点，维护文件系统树和以及树中的文件和目录。说白了，就是一个目录。 但是这个目录有非常的重要，里面存放了命名空间的镜像文件和编辑日志文件，用户拼接数据节点的信息（存放目录 /Users/zzjmay/hadoopwork/hadoop_tmp/dfs/name/）
datanode：数据节点，用于存放数据块的节点(存放目录/Users/zzjmay/hadoopwork/hadoop_tmp/dfs/data/ )
注意：
1. 正常开发中，名称节点和数据节点是要分开部署的
2. 名称节点要做好容灾措施，两种方式：第一种，配置远程的文件系统，以保证一致性 第二种，创建辅助的名称节点

回到之前说的pycharm的分割的数据块存放的路径
/Users/zzjmay/hadoopwork/hadoop_tmp/dfs/data/current/BP-1423851909-127.0.0.1-1522918427220/current/finalized/subdir0/subdir0

Hadoop的采用的是java的切割技术，只需要将这三个数据块拼接在一起就可以回到原来的文件，亲测有效

namenode只存放元数据信息，不会存储块信息。当datanode启动后，会主动请求namenode，去报备对应的快列表

Java 操作HDFS文件系统步骤（Maven模式下）

步骤1 创建maven项目（不说）

步骤2 引入hadoop的依赖

 <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.7.3</version>
  </dependency>

步骤3 写代码了

先介绍几个重要的类

FileSystem

定义了Hadoop的文件系统的API接口，提供了多种构造函数

Configuration

这个类的作用有两个，1. 读取core-site类文件和设置相关配置信息 2. 后面会用到它来传递参数

FSDataInputStream

一般通过FileSystem.open（Path）获取. hdfs的输入流

FSDataOutputStream

HDFS的输出流，比如向创建一个文件FileSystem.create就会产生这个流

FileStatus

存放的是Hdfs文件目录或文件的相关信息

按照下面的代码，对于Hdfs的操作你就可以基本掌握了

package com.zzjmay.hdfsapi;

import com.zzjmay.utils.ConfUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
import org.junit.Test;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URI;
import java.net.URL;
import java.net.URLConnection;

/**
 * 测试Hdfs Api
 * Created by zzjmay on 2018/4/7.
 */
public class TestHdfsApi {
    /**
     * 这么做的原因是java程序是不识别hdfs协议的，通过加载这个方法让URL识别
     */
    static {
        URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
    }

    /**
     * 读取hdfs文件 URL读取数据
     */
    @Test
    public void readFileByURL() {

        //JDK7以上的写法，IO流在try中会自动关闭
        try (InputStream inputStream = new URL("hdfs://localhost:8000/zzjmay/hadoop/sumApp.txt").openStream();) {

            IOUtils.copyBytes(inputStream, System.out, 4096, false);
        } catch (Exception e) {
            System.out.println("##出错了");
        }

    }

    /**
     * 通过Hadoop API访问文件
     * 相较于第一种URL的方式，这种方式更常用
     * FileSystem API 来打开一个文件的输入流
     */
    @Test
    public void readFileByAPI() {
        //1.封装了客户端或服务器端的配置
        Configuration conf = new Configuration();
        FSDataInputStream inputStream = null;
        try {
            //书本上的方式，配置URI和conf
            //2.通过fileSystem的静态方法get获取FileSystem 对象
            FileSystem fileSystem = FileSystem.get(URI.create("hdfs://127.0.0.1:8000/user/zzjmay/笔记.txt"), conf);
            //3.获取hdfs逻辑目录下的文件
            inputStream = fileSystem.open(new Path("/user/zzjmay/笔记.txt"));
            //4.读取文件输入流
            IOUtils.copyBytes(inputStream, System.out, 4096, false);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            IOUtils.closeStream(inputStream);
        }


    }

    @Test
    public void readFileByAPI2() {
        //1.封装了客户端或服务器端的配置
        Configuration conf = new Configuration();
        //因为目前采用的是伪分布模式,需要设置读取的名称节点的主机名,这里要注意不能用localhost
        //在hadoop中，fs.default.name 已经替换为fs.defaultFS
        conf.set("fs.defaultFS", "hdfs://127.0.0.1:8000");
        FSDataInputStream inputStream = null;
        try {
            FileSystem fileSystem = FileSystem.get(conf);
            //3.获取hdfs逻辑目录下的文件
            inputStream = fileSystem.open(new Path("/user/zzjmay/笔记.txt"));
            //4.读取文件输入流
            IOUtils.copyBytes(inputStream, System.out, 4096, false);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            IOUtils.closeStream(inputStream);
        }
    }

    /**
     * 在HDFS中创建目录
     */
    @Test
    public void mkdir() {
        //1.加载conf基础配置
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://127.0.0.1:8000");

        try {
            //2。获取操作HDFS的API
            FileSystem fileSystem = FileSystem.get(conf);
            //3.获取FS输出流
            fileSystem.mkdirs(new Path("/user/zzjmay/hadoop"));

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

    /**
     * 写入文件
     */
    @Test
    public void putFile() {
        FSDataOutputStream outputStream = null;
        try {
            FileSystem fileSystem = FileSystem.get(ConfUtils.getConf());
            outputStream = fileSystem.create(new Path("/user/zzjmay/hadoop/test.txt"));

            outputStream.write("helloword".getBytes());

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            IOUtils.closeStream(outputStream);
        }
    }

    /**
     * 删除文件
     */
    @Test
    public void delteFile() {

        try {
            FileSystem fileSystem = FileSystem.get(ConfUtils.getConf());
            fileSystem.delete(new Path("/zzjmay/hadoop/pycharm-professional-2016.3.1.dmg"),true);

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

    /**
     * 从本地复制文件当hdfs上
     * 并且显示进度
     */
    @Test
    public void testProcess(){
        InputStream inputStream = null;
        try {
            //1.读取本地文件
             inputStream = new BufferedInputStream(new FileInputStream("/Users/zzjmay/hadoopwork/data/test.txt"));
             //2. 获取操作hdfs操作流
            FileSystem fileSystem = FileSystem.get(ConfUtils.getConf());
            FSDataOutputStream outputStream = fileSystem.create(new Path("/user/zzjmay/process.txt"), new Progressable() {
                @Override
                public void progress() {
                    System.out.println("####测试哈哈哈");
                }
            });
            IOUtils.copyBytes(inputStream,outputStream,4096,true);
        }catch (Exception e){

        }

    }
    @Test
    public void getFileStatus(){
        try {
            FileSystem fileSystem = FileSystem.get(ConfUtils.getConf());
            //获取文件的信息FileStauts
            FileStatus fileStatus = fileSystem.getFileStatus(new Path("/user/zzjmay/process.txt"));
            System.out.println("块大小"+fileStatus.getBlockSize());
        }catch (Exception e){

        }

    }


}