精选文章 分布式文件系统HDFS(三)

分布式文件系统HDFS(三)

作者:爱学习的小肥猪 时间: 2019-11-07 03:37:28
爱学习的小肥猪 2019-11-07 03:37:28

7、HDFS的文件写入过程

分布式文件系统HDFS(三)1 
 

详细步骤解析:
1、 client发起文件上传请求,通过RPC与NameNode建立通讯,NameNode检查目标文件是否已存在,父目录是否存在,返回是否可以上传;
2、 client请求第一个block该传输到哪些DataNode服务器上;
3、 NameNode根据配置文件中指定的备份数量及机架感知原理进行文件分配,返回可用的DataNode的地址如:A,B,C;
 
注:Hadoop在设计时考虑到数据的安全与高效,数据文件默认在HDFS上存放三份,存储策略为本地一份,同机架内其它某一节点上一份,不同机架的某一节点上一份。
4、 client请求3台DataNode中的一台A上传数据(本质上是一个RPC调用,建立pipeline),A收到请求会继续调用B,然后B调用C,将整个pipeline建立完成,后逐级返回client;
5、 client开始往A上传第一个block(先从磁盘读取数据放到一个本地内存缓存),以packet为单位(默认64K),A收到一个packet就会传给B,B传给C;A每传一个packet会放入一个应答队列等待应答。
6、 数据被分割成一个个packet数据包在pipeline上依次传输,在pipeline反方向上,逐个发送ack(命令正确应答),最终由pipeline中第一个DataNode节点A将pipelineack发送给client;
7、 当一个block传输完成之后,client再次请求NameNode上传第二个block到服务器。
 

8、HDFS的文件读取过程



分布式文件系统HDFS(三)2
 

详细步骤解析
1、 Client向NameNode发起RPC请求,来确定请求文件block所在的位置;
2、 NameNode会视情况返回文件的部分或者全部block列表,对于每个block,NameNode 都会返回含有该 block副本的 DataNode 地址;  这些返回的 DN 地址,会按照集群拓扑结构得出 DataNode 与客户端的距离,然后进行排序,排序两个规则:网络拓扑结构中距离 Client 近的排靠前;心跳机制中超时汇报的 DN 状态为 STALE,这样的排靠后;
3、 Client 选取排序靠前的DataNode 来读取 block,如果客户端本身就是DataNode,那么将从本地直接获取数据(短路读取特性);
4、 底层上本质是建立 Socket Stream(FSDataInputStream),重复的调用父类DataInputStream 的 read 方法,直到这个块上的数据读取完毕;
5、 当读完列表的 block 后,若文件读取还没有结束,客户端会继续向NameNode 获取下一批的 block 列表;
6、 读取完一个 block 都会进行 checksum验证,如果读取 DataNode 时出现错误,客户端会通知 NameNode,然后再从下一个拥有该 block 副本的DataNode 继续读。
7、 read 方法是并行的读取 block 信息,不是一块一块的读取;NameNode只是返回Client请求包含块的DataNode地址,并不是返回请求块的数据;
8、 最终读取来所有的 block 会合并成一个完整的最终文件。
 
9、HDFS的API操作9.1、创建maven工程并导入jar包
由于cdh版本的所有的软件涉及版权的问题,所以并没有将所有的jar包托管到maven仓库当中去,而是托管在了CDH自己的服务器上面,所以我们默认去maven的仓库下载不到,需要自己手动的添加repository去CDH仓库进行下载,以下两个地址是官方文档说明,请仔细查阅
https://www.cloudera.com/documentation/enterprise/release-notes/topics/cdh_vd_cdh5_maven_repo.html
https://www.cloudera.com/documentation/enterprise/release-notes/topics/cdh_vd_cdh5_maven_repo_514x.html
 
   
       cloudera
       https://repository.cloudera.com/artifactory/cloudera-repos/
   
   
       org.apache.hadoop
       hadoop-client
       2.6.0-mr1-cdh5.14.0
   
   
       org.apache.hadoop
       hadoop-common
       2.6.0-cdh5.14.0
   
   
       org.apache.hadoop
       hadoop-hdfs
       2.6.0-cdh5.14.0
   
 
   
       org.apache.hadoop
       hadoop-mapreduce-client-core
       2.6.0-cdh5.14.0
   
   
   
       junit
       junit
       4.11
       test
   
   
       org.testng
       testng
       RELEASE
   
   
      
           org.apache.maven.plugins
           maven-compiler-plugin
           3.0
           
               1.8
               1.8
               UTF-8
               
           
      
 
      
           org.apache.maven.plugins
            maven-shade-plugin
           2.4.3
           
               
                   package
                  
                       shade
                  
                  
                       true
                  
               
           
      
     
   

9.2、使用url的方式访问数据(了解)
@Test
public void demo1()throws  Exception{
    //第一步:注册hdfs url,让java代码能够识别hdfsurl形式
    
URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());

    InputStream inputStream = null;
    FileOutputStream outputStream =null;
    //定义文件访问的url地址
    
Stringurl = "hdfs://192.168.52.100:8020/test/input/install.log";
    //打开文件输入流
    
try {
        inputStream = new URL(url).openStream();
        outputStream = new FileOutputStream(newFile("c:\\hello.txt"));
        IOUtils.copy(inputStream,outputStream);
    } catch (IOExceptione) {
        e.printStackTrace();
    }finally {
        IOUtils.closeQuietly(inputStream);
        IOUtils.closeQuietly(outputStream);
    }
}
 
 
如果执行出现以下错误,可以参见资料如何解决,也可以不用理会,不会影响程序的执行。记得配置完成环境变量之后重启开发工具
                              
 
9.3、使用文件系统方式访问数据(掌握)
 
在 java 中操作 HDFS,主要涉及以下 Class:
Configuration:该类的对象封转了客户端或者服务器的配置; FileSystem:该类的对象是一个文件系统对象,可以用该对象的一些方法来对文件进行操作,通过 FileSystem 的静态方法 get 获得该对象。
FileSystem fs = FileSystem.get(conf)
get 方法从 conf 中的一个参数fs.defaultFS 的配置值判断具体是什么类型的文件系统。如果我们的代码中没有指定fs.defaultFS,并且工程 classpath下也没有给定相应的配置,conf中的默认值就来自于hadoop的jar包中的core-default.xml ,默 认 值 为 : file:/// , 则 获 取 的 将 不 是 一 个DistributedFileSystem的实例,而是一个本地文件系统的客户端对象
 
 
9.4、获取FileSystem的几种方式
第一种方式获取FileSystem
@Test
public void getFileSystem() throws URISyntaxException,IOException {
   Configuration configuration = new Configuration();
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"),configuration);
    System.out.println(fileSystem.toString());
}
 
第二种获取FileSystem类的方式
@Test
public void getFileSystem2() throws URISyntaxException,IOException {
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS","hdfs://192.168.52.100:8020");
    FileSystem fileSystem = FileSystem.get(new URI("/"),configuration);
    System.out.println(fileSystem.toString());
}
 
第三种获取FileSystem类的方式
@Test
public void getFileSystem3() throws URISyntaxException,IOException {
    Configuration configuration = new Configuration();
    FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://192.168.52.100:8020"),configuration);
    System.out.println(fileSystem.toString());
}
 
 
第四种获取FileSystem类的方式
@Test
public void getFileSystem4() throws  Exception{
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS","hdfs://192.168.52.100:8020");
    FileSystem fileSystem = FileSystem.newInstance(configuration);
    System.out.println(fileSystem.toString());
}
 
9.5、递归遍历文件系统当中的所有文件
通过递归遍历hdfs文件系统
@Test
public void listFile() throws Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"),new Configuration());
    FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/"));
    for (FileStatusfileStatus : fileStatuses) {
        if(fileStatus.isDirectory()){
            Path path =fileStatus.getPath();
           listAllFiles(fileSystem,path);
        }else{
            System.out.println("文件路径为"+fileStatus.getPath().toString());

        }
    }
}
public void listAllFiles(FileSystemfileSystem,Path path) throws  Exception{
    FileStatus[] fileStatuses =fileSystem.listStatus(path);
    for (FileStatusfileStatus : fileStatuses) {
        if(fileStatus.isDirectory()){
            listAllFiles(fileSystem,fileStatus.getPath());
        }else{
            Path path1 =fileStatus.getPath();
            System.out.println("文件路径为"+path1);
        }
    }
}
官方提供的API直接遍历
/**
递归遍历官方提供的API版本
@throws Exception
*/

@Test
public void listMyFiles()throws Exception{
    
//获取fileSystem
    
FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"), newConfiguration());
    
//获取RemoteIterator 得到所有的文件或者文件夹,第一个参数指定遍历的路径,第二个参数表示是否要递归遍历
    
RemoteIteratorlocatedFileStatusRemoteIterator = fileSystem.listFiles(new Path("/"), true);
    while (locatedFileStatusRemoteIterator.hasNext()){
        LocatedFileStatus next =locatedFileStatusRemoteIterator.next();
        System.
out.println(next.getPath().toString());
    }
   fileSystem.close();
}
 
9.6、下载文件到本地
程序执行的main方法
/**
拷贝文件的到本地
@throws Exception
*/

@Test
public void getFileToLocal()throws  Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"),new Configuration());
    FSDataInputStream open = fileSystem.open(new Path("/test/input/install.log"));
    FileOutputStream fileOutputStream = new FileOutputStream(newFile("c:\\install.log"));
    IOUtils.copy(open,fileOutputStream);
    IOUtils.closeQuietly(open);
    IOUtils.closeQuietly(fileOutputStream);
   fileSystem.close();
}
 
 
9.7、hdfs上创建文件夹
@Test
public void mkdirs() throws  Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"),new Configuration());
    boolean mkdirs= fileSystem.mkdirs(new Path("/hello/mydir/test"));
    fileSystem.close();
}
9.8、hdfs文件上传
@Test
public void putData() throws  Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"),new Configuration());
    fileSystem.copyFromLocalFile(new Path("file:///c:\\install.log"),new Path("/hello/mydir/test"));
    fileSystem.close();
}
 
 
 
9.9、HDFS权限问题以及伪造用户
首先停止hdfs集群,在node01机器上执行以下命令
cd /export/servers/hadoop-2.6.0-cdh5.14.0
sbin/stop-dfs.sh
修改node01机器上的hdfs-site.xml当中的配置文件
cd/export/servers/hadoop-2.6.0-cdh5.14.0/etc/hadoop
vim hdfs-site.xml
               dfs.permissions
               true
      
 
修改完成之后配置文件发送到其他机器上面去
scp hdfs-site.xml node02:$PWD
scp hdfs-site.xml node03:$PWD
重启hdfs集群
cd /export/servers/hadoop-2.6.0-cdh5.14.0
sbin/start-dfs.sh
 
随意上传一些文件到我们hadoop集群当中准备测试使用
cd/export/servers/hadoop-2.6.0-cdh5.14.0/etc/hadoop
hdfs dfs -mkdir /config
hdfs dfs -put *.xml /config
hdfs dfs -chmod 600 /config/core-site.xml   
 
使用代码准备下载文件
@Test
public void getConfig()throws  Exception{
    FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"),new Configuration(),"[url=]root[/url][a1] ");
    fileSystem.copyToLocalFile(new Path("/config/core-site.xml"),new Path("file:///c:/core-site.xml"));
    fileSystem.close();
}
 
9.10、HDFS的小文件合并
由于hadoop擅长存储大文件,因为大文件的元数据信息比较少,如果hadoop集群当中有大量的小文件,那么每个小文件都需要维护一份元数据信息,会大大的增加集群管理元数据的内存压力,所以在实际工作当中,如果有必要一定要将小文件合并成大文件进行一起处理
在我们的hdfs 的shell命令模式下,可以通过命令行将很多的hdfs文件合并成一个大文件下载到本地,命令如下
cd /export/servers
hdfs dfs -getmerge /config/*.xml  ./hello.xml
 
既然可以在下载的时候将这些小文件合并成一个大文件一起下载,那么肯定就可以在上传的时候将小文件合并到一个大文件里面去
代码如下:
 
/**
将多个本地系统文件,上传到hdfs,并合并成一个大的文件
@throws Exception
*/

@Test
public void mergeFile() throws  Exception{
    //获取分布式文件系统
    
FileSystemfileSystem = FileSystem.get(new URI("hdfs://192.168.52.100:8020"), new Configuration(),"root");
    FSDataOutputStream outputStream =fileSystem.create(new Path("/bigfile.xml"));
    //获取本地文件系统
    
LocalFileSystemlocal = FileSystem.getLocal(new Configuration());
    //通过本地文件系统获取文件列表,为一个集合
    
FileStatus[]fileStatuses = local.listStatus(new Path("file:///F:\\传智播客大数据离线阶段课程资料\\3、大数据离线第三天\\上传小文件合并"));
    for (FileStatusfileStatus : fileStatuses) {
        FSDataInputStream inputStream =local.open(fileStatus.getPath());
       IOUtils.copy(inputStream,outputStream);
        IOUtils.closeQuietly(inputStream);
    }
    IOUtils.closeQuietly(outputStream);
    local.close();
    fileSystem.close();
}

 
勿删,copyright占位
分享文章到微博
分享文章到朋友圈

上一篇:EasyUI入门

下一篇:SQL 2005 权限控制粒度

您可能感兴趣

  • 二本学历,五年抄代码经验,疫情期被裁,三个月25K入职字节跳动

    前言 我的个人背景非常简单,也可以说丝毫没有亮点。 学历普通,计算机专业二本毕业,毕业后出来就一直在一家小公司,岁月如梭细,算了下至今从事Java开发已经5年了,也crud了五年,每天就是抄代码,毫无新意,甚至一度的怀疑人生。 其实很久之前就已经对自己不满了,每次都在想鼓起勇气想去打破困境,去改变自己,然而因为游戏因为玩耍一次又一次的破功;直到这次疫情,强有力的给了我当头一棒。是的,我失业了...

  • 四连涨!海尔智家这波私有化还有多大想象空间?

    文|易牟 来源 | 螳螂财经(ID:TanglangFin) 四连涨!受海尔智家(600690.SH)私有化海尔电器(01169.HK)消息影响,海尔智家连续两个交易日涨停后,又接连两日上攻,股价盘中最高达22.54元,创下今年新高。 截至8月5日收盘,海尔智家主力资金净流入3.68亿元,连续5日净流入12.96亿元,后续上涨势头强劲。 毫无疑问,在投资者的眼中,不管是短期还是中长期,海尔智...

  • 功能fine秒mine!苹果“偷走”了安卓系统,但它表现得更好

    全文共1670字,预计学习时长5分钟 图源:unsplash 不久前的开发者大会上,苹果发布了iOS 14,展示了一些将在今年晚些时候出现在iPhone上的重要功能,比如允许用户将应用程序中的内容导入主屏幕的小部件,以及画中画视频功能。 安卓迷们发现了一些很明显的事情:许多在年度开发者大会上发布的功能已经在安卓上使用了很多年了。没错,小部件从一开始就是安卓系统的一部分,它可以让用户在主屏幕上...

  • 爱了,3174页实战pdf集锦:Redis+多线程+Dubbo+JVM+kafka+MySQL

    写在前面 作为一名Java开发者,在现在这个信息化时代很快的时代,很少会有人停下脚步去思考以及去总结,忽略了很重要的一个步骤,没有反思和总结,只会用原来固有的想法去做事情,所以还是需要隔一段时间去总结。LZ今天总结了自己在平时会用到的一些: 01—Redis实战 在Redis诞生数年之后的今天,这个项目已经发生了显著的变化:我们现在拥有了一个更为健壮的系统,并且随着Redis 2.6的发布,...

  • 【计算机网络】第七章:网络安全

    【计算机网络】第七章:网络安全 目录 网络安全问题概述 1.1.计算机网络面临的安全性威胁 1.2.被动攻击和主动攻击 1.3.计算机网络通信安全的目标 1.4.恶意程序(rogue program) 1.5.计算机网络安全的内容 1.6.一般的数据加密模型 两类密码体制 一、对称密钥密码体制 1.1.对称密钥的缺点与优点 1.2.数据加密标准 DES 1.3.DES 的保密性 二、公钥密码...

  • Redis到底该怎么搞呢?主从复制以及redis复制演进全面进攻

    一、前言 通过持久化功能,Redis保证了即使在服务器宕机情况下数据的丢失非常少。但是如果这台服务器出现了硬盘故障、系统崩溃等等,不仅仅是数据丢失,很可能对业务造成灾难性打击。为了避免单点故障通常的做法是将数据复制多个副本保存在不同的服务器上,这样即使有其中一台服务器出现故障,其他服务器依然可以继续提供服务。当然Redis提供了多种高可用方案包括:主从复制、哨兵模式的主从复制、以及集群。 本...

  • Web端即时通讯实践干货:如何让WebSocket断网重连更快速?

    本文作者网易智慧企业web前端开发工程师马莹莹。为了提升内容质量,收录时有修订和改动。 1、引言 在一个完善的即时通讯IM应用中,WebSocket是极其关键的一环,它为基于Web的即时通讯应用提供了一种全双工的通信机制。但为了提升IM等实际应用场景下的消息即时性和可靠性,我们需要克服WebSocket及其底层依赖的TCP连接对于复杂网络情况下的不稳定性,即时通讯的开发者们通常都需要为其设计...

  • 工作流学习2(书本)

    1、流程引擎的创建。 1.1、ProcessEngineConfiguration的buildProcessEngine方法 使用ProcessEngineConfiguration的create方法可以得到ProcessEngineConfiguration的实例。ProcessEngineConfiguration中提供了一个buildProcessEngine方法,该方法返回一个Pro...

华为云40多款云服务产品0元试用活动

免费套餐,马上领取!
CSDN

CSDN

中国开发者社区CSDN (Chinese Software Developer Network) 创立于1999年,致力为中国开发者提供知识传播、在线学习、职业发展等全生命周期服务。