DOIT20-HDP02

2020/12/18 11:49
阅读数 38

1 HDFS的shell客户端

[root@linux01 ~]# hdfs  dfs  
Usage: hadoop fs [generic options]
        [-appendToFile <localsrc> ... <dst>]
        [-cat [-ignoreCrc] <src> ...]
        [-checksum <src> ...]
        [-chgrp [-R] GROUP PATH...]
        [-chmod [-R] <MODE[,MODE]... | OCTALMODE> PATH...]
        [-chown [-R] [OWNER][:[GROUP]] PATH...]
        [-copyFromLocal [-f] [-p] [-l] [-d] [-t <thread count>] <localsrc> ... <dst>]
        [-copyToLocal [-f] [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
        [-count [-q] [-h] [-v] [-t [<storage type>]] [-u] [-x] [-e] <path> ...]
        [-cp [-f] [-p | -p[topax]] [-d] <src> ... <dst>]
        [-createSnapshot <snapshotDir> [<snapshotName>]]
        [-deleteSnapshot <snapshotDir> <snapshotName>]
        [-df [-h] [<path> ...]]
        [-du [-s] [-h] [-v] [-x] <path> ...]

        [-expunge]
        [-find <path> ... <expression> ...]
        [-get [-f] [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
        [-getfacl [-R] <path>]
        [-getfattr [-R] {-n name | -d} [-e en] <path>]
        [-getmerge [-nl] [-skip-empty-file] <src> <localdst>]
        [-head <file>]
        [-help [cmd ...]]
        [-ls [-C] [-d] [-h] [-q] [-R] [-t] [-S] [-r] [-u] [-e] [<path> ...]]
        [-mkdir [-p] <path> ...]
        [-moveFromLocal <localsrc> ... <dst>]
        [-moveToLocal <src> <localdst>]
        [-mv <src> ... <dst>]
        [-put [-f] [-p] [-l] [-d] <localsrc> ... <dst>]
        [-renameSnapshot <snapshotDir> <oldName> <newName>]
        [-rm [-f] [-r|-R] [-skipTrash] [-safely] <src> ...]
        [-rmdir [--ignore-fail-on-non-empty] <dir> ...]

        [-setfacl [-R] [{-b|-k} {-m|-x <acl_spec>} <path>]|[--set <acl_spec> <path>]]
        [-setfattr {-n name [-v value] | -x name} <path>]
        [-setrep [-R] [-w] <rep> <path> ...]
        [-stat [format] <path> ...]
        [-tail [-f] <file>]
        [-test -[defsz] <path>]
        [-text [-ignoreCrc] <src> ...]
        [-touchz <path> ...]
        [-truncate [-w] <length> <path> ...]
        [-usage [cmd ...]]







































hdfs  dfs  -ls   /
hdfs  dfs  -put  /a.sh    /data/
hdfs  dfs  -get  /data/a.sh   /doit20/
hdfs  dfs  -cat  /a.sh
hdfs  dfs  -tail /a.sh
hdfs  dfs  -head /a.sh
hdfs  dfs  -chmod  -R  777   /data
hdfs  dfs  -mv   
hdfs  dfs  -cp
hdfs  dfs  -mkdir  -p 
hdfs  dfs  -rm  -r 
hdfs  dfs  -rmdir 
hdfs  dfs  -df -h
hdfs  dfs  -du  -h
hdfs  dfs  -find  /data/  -name  a.sh

2 HDFS的java客户端

1 maven项目

 <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>
 <!--项目需要的jar包   mysql   fastjson...-->
    <dependencies>
        <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.47</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.62</version>
        </dependency>
        <!--添加hdfs的客户端依赖-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>3.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>3.1.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-common</artifactId>
            <version>3.1.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
            <version>3.1.1</version>
        </dependency>

    </dependencies>

</project>

2 编程

2.1 入门程序

package cn._51doit.day02.hdfs.client;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.net.URI;
import java.net.URISyntaxException;

/**
 * FileName: Demo1
 * Author:   多易教育-DOIT
 * Date:     2020/12/16 0016
 * Description:使用java操作HDFS分布式文件系统
 */
public class Demo1 {
    public static void main(String[] args) throws Exception {
        // 1 配置信息对象    用于用户自定义设置 比如副本个数   物理切块
        Configuration conf = new Configuration();
        // 2 获取HDFS 的客户端对象
        /**
         * newInstance获取文件系统
         * 参数一  URI   namenode的位置  HDFS的位置
         * URL http://linux01:8080/html/login.html
         * jdbc:mysql://localhost:3306/db_dmeo1
         * hdfs://linux01:8020
         * 参数二 配置对象
         * 参数三  用户名
         */
        FileSystem fs = FileSystem.newInstance(new URI("hdfs://linux01:8020"), conf, "root");
         // 操作HDFS 系统  ls put get mv   rm   cat
        // 将windows中的文件上传到HDFS 中
        /**
         * 参数一 本地路径
         * 参数二  HDFS 的路径
         */
        fs.copyFromLocalFile(new Path("d://word.txt"),new Path("/"));

        fs.close();

    }
}

2.2 上传下载

    /**
     * 获取HDFS客户端
     * hdfs://linux01:8020
     * @return
     * @throws Exception
     */
    public  static FileSystem getHdfsFs() throws Exception {
        Configuration conf = new Configuration();
        return  FileSystem.newInstance(new URI("hdfs://linux01:8020"),conf , "root");

    }
 /**
     * 下载
     * @throws Exception
     */
    private static void get() throws Exception {
        FileSystem fs = HdfsUtils.getHdfsFs();
        /**
         * 注意:************ 下载API需要windows本地配置HDP环境变量  且生效
         * 参数一  是否删除待下载的文件
         * 参数二  HDFS的路径
         * 参数三 本地 路径
         * 参数四  是否使用本地原生的文件系统 ***  默认是false
         *     是否使用原生的本地文件系统  true使用windows的文件系统来存储存储下载的数据  所以在本地没有生成.word.txt.crc文件
         *     默认使用HDFS系统 ---> windows写数据 生成校验文件 .a.sh.crc
         */
        fs.copyToLocalFile(true , new Path("/a.sh") ,new Path("d://"),false);
        fs.close();
    }

    /**
     * 上传
     * @throws Exception
     */
    private static void put() throws Exception {
        //1 上传
        FileSystem fs = HdfsUtils.getHdfsFs();
        /**
         * 参数一  是否删除待上传的文件
         * 参数二  是否覆盖HDFS中已经存在的文件
         * 参数三  待上传文件路径
         * 参数四  HDFS的目标路径
         */
        fs.copyFromLocalFile(true, true, new Path("d://word.txt"), new Path("/"));
        fs.close();
    }

2.3 创建文件夹

 /**
     * 创建文件夹
     *
     * @throws Exception
     */
    private static void mkdirs() throws Exception {
        FileSystem fs = HdfsUtils.getHdfsFs();
        //创建多级文件夹
        boolean b = fs.mkdirs(new Path("/wbb/wb/b"));
        // 如果创建成功返回true
        if (b) {// 成功
            // 遍历/下所有的内容[FileStatus 包括文件和文件夹]
            FileStatus[] statuses = fs.listStatus(new Path("/"));
            for (FileStatus status : statuses) {
                // 获取文件文件夹 路径
                Path path = status.getPath();
                // 获取名称
                String name = path.getName();
                System.out.println(name);
            }
        }
        fs.close();
    }

2.4 删除内容

  /**
     * 删除内容
     *
     * @throws Exception
     */
    private static void rmr() throws Exception {
        FileSystem fs = HdfsUtils.getHdfsFs();
        boolean b = fs.delete(new Path("/wbb"), true);
        if (b) {
            System.out.println("删除成功");
        } else {
            System.out.println("删除失败");
        }
        fs.close();
    }

2.5 遍历路径下所有的文件

  /**
     * 遍历路径下所有的文件
     *
     * @throws Exception
     */
    private static void listFiles() throws Exception {
        FileSystem fs = HdfsUtils.getHdfsFs();
        // 递归遍历一个路径下所有的文件
        RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fs.listFiles(new Path("/"), true);
        // 遍历文件
        while (locatedFileStatusRemoteIterator.hasNext()) {
            // 获取每个文件
            LocatedFileStatus next = locatedFileStatusRemoteIterator.next();
            // 获取每个文件的路径
            Path path = next.getPath();
            System.out.println(path);
            String name = path.getName();  // 获取每个文件的文件名
        }
        fs.close();
    }
hdfs://linux01:8020/a/b/c/zk.sh
hdfs://linux01:8020/a/b/zk.sh
hdfs://linux01:8020/wbb.mp4

2.6 API01

    /**
     * 读取数据
     * @throws Exception
     */
    private static void readData() throws Exception {
        FileSystem fs = HdfsUtils.getHdfsFs();
        // 8 读取数据
        FSDataInputStream  ipt = fs.open(new Path("/ab.mp4"));
        //wbbhefengjiede qimei aiqing gushi
      /*  int i = ipt.read();
        int j = ipt.read();
        System.out.println(j);*/
        BufferedReader br = new BufferedReader(new InputStreamReader(ipt));
        String  line = null ;
        while((line = br.readLine())!=null){
            System.out.println(line);
        }
        ipt.close();
        br.close();
        fs.close();
    }

    /**
     * 移动 重命名
     * @param fs
     * @throws IOException
     */
    private static void mv(FileSystem fs) throws IOException {
        fs.rename(new Path("/a.mp4") , new Path("/ab.mp4")) ;
    }

    /**
     * 获取元数据信息
     * @throws Exception
     */
    private static void getFileBlockMeta() throws Exception {
        FileSystem fs = HdfsUtils.getHdfsFs();
        // 递归遍历一个路径下所有的文件
        RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fs.listFiles(new Path("/"), false);
        // 遍历文件
        while (locatedFileStatusRemoteIterator.hasNext()) {
            // 获取每个文件
            LocatedFileStatus next = locatedFileStatusRemoteIterator.next();
            // 获取每个文件的路径
            Path path = next.getPath();
            System.out.println(path);
            String name = path.getName();  // 获取每个文件的文件名
            next.getBlockSize() ;//128M
            next.getReplication() ;//3
            //  获取元数据信息..
            BlockLocation[] blockLocations = next.getBlockLocations();// 块位置   a.txt  300M
            for (BlockLocation blockLocation : blockLocations) {//块   3个副本
                long length = blockLocation.getLength(); // 数据块的大小
                String[] names = blockLocation.getNames();
                System.out.println(Arrays.asList(names));// names
                // 获取乜咯副本所在的机器
                String[] hosts = blockLocation.getHosts(); // 3
                List<String> strings = Arrays.asList(hosts); // hosts
                System.out.println(strings);
            }
        }
        fs.close();
    }

    /**
     * 遍历路径下所有的内容
     * @throws Exception
     */
    private static void listStatus() throws Exception {
        FileSystem fs = HdfsUtils.getHdfsFs();
        FileStatus[] fileStatuses = fs.listStatus(new Path("/"));
        for (FileStatus fileStatus : fileStatuses) {
            fileStatus.isDirectory();
            if(fileStatus.isFile()){  // 文件
              //  获取文件信息
                fileStatus.getAccessTime() ;
                fileStatus.getBlockSize();
                fileStatus.getLen() ;
                fileStatus.getReplication() ;
            }
        }
        fs.close();
    }

2,7 追加写

 /**
     * 追加写操作
     * @throws Exception
     */
    private static void writeData() throws Exception {
        FileSystem fs = HdfsUtils.getHdfsFs();
        FSDataOutputStream append = fs.append(new Path("/ab.mp4"));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(append));
        bw.write("wbb 真笨");
        bw.newLine();
        bw.flush();
        bw.close();
        fs.close();
    }

3 原理加强

3.1 数据存储位置观察

当datanode启动以后会向namenode注册 ,dfs.namenode.rpc-address ! namenode接收注册 返回一个唯一的集群ID!

返回一个blockpoolId, datanode将数据存储在这个统一的目录中!

-- 数据在集群中是以128M物理切块多副本存储 (集群),数据块回存储在

每台datanode的这个BP-859207207-192.168.133.3-1608020883489目录下

在linux01机器  linux02 linux03的目录下有数据块

3.2 写数据流程(上传)

....作业

 

 

3.3 读数据流程 (下载)

 

3.4 namenode和datanode通信

1、namenode 如何判断datanode节点是否宕机?
先决条件:
datanode每隔一段时间像namenode汇报,汇报的信息有两点
(1)自身datanode的状态信息;
(2)自身datanode所持有的所有的数据块的信息。
如果namenode连续十次没有收到datanode的汇报,那么namenode就会认为该datanode存在宕机的可能。
datanode启动以后会专门启动一个进程负责给namenode发送心跳数据包,如果datanode没有问题,仅仅只是发送信息数据包的进程挂了,namenode会发送命令像这个datanode进行确认,
查看这个发送心跳包的进程是否还能正常运行,namenode会向datanode确认两遍,每五分钟确认一次。如果两次都没有返回结果,那么namenode就会认为datanode已经宕机了。
最终namenode判断一个datanode死亡的时间计算公式:
timeout = 10 * 心跳间隔时间 + 2 * 检查一次消耗的时间
心跳间隔时间
配置参数:dfs.heartbeat.interval
默认值:3s,单位s
检查一次消耗的时间
配置参数:dfs.namenode.heartbeat.recheck-interval
默认值:    300000,单位ms

2、数据块读写操作的一些配置项
(1)配置参数:dfs.blocksize 
参数说明:客户端数据分块的大小
默认值:134217728(128M);单位:字节
(2)dfs.client-write-packet-size
参数说明:客户端写入数据时packet数据包的大小
默认值:65536(64K),单位字节
(3)dfs.bytes-per-checksum
参数说明:chunk(写入数据时最小的一个单位),数据校验的粒度
默认值:512字节
注:事实上一个chunk还包含4B的校验值,因而chunk写入packet时是516B;数据与检验值的比值为128:1,所以对于一个128M的block会有一个1M的校验文件与之对应;

3、集群数据损坏以后,自动删除时长
配置项:dfs.blockreport.intervalMsec
默认值:21600000(36分钟),单位:ms

4、元数据checkPount
在分布式或者伪分布集群中,每隔一段时间,会由 secondary namenode 将 namenode 上积累的所有 edits 和一个最新的 fsimage 下载到本地,并加载到内存进行 merge(这个过程称为 checkpoint)
dfs.namenode.checkpoint.check.period=60 ##检查触发条件是否满足的频率,60 秒
dfs.namenode.checkpoint.dir=file://${hadoop.tmp.dir}/dfs/namesecondary
##以上两个参数做 checkpoint 操作时,secondary namenode 的本地工作目录
dfs.namenode.checkpoint.edits.dir=${dfs.namenode.checkpoint.dir}
dfs.namenode.checkpoint.max-retries=3 ##最大重试次数
dfs.namenode.checkpoint.period=3600 ##两次 checkpoint 之间的时间间隔 3600 秒
dfs.namenode.checkpoint.txns=1000000 ##两次 checkpoint 之间最大的操作记录

5、jvm重用
配置参数:mapred.job.reuse.jvm.num.tasks
参数解释:一个jvm内部跑多少个task,默认是1,可以设置为多个。这个参数在2.9.2版本中未找到。

 

3.5  namenode和datanode角色总结 

3.6 元数据管理机制

 

 

 

 

 

 

 

 

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部