淘先锋技术网

首页 1 2 3 4 5 6 7

hadoop---HDFS核心概念以及HDFS Java Api的使用

HDFS概念和存储机制：

   概念: 分布式文件系统，主要是文件系统。
   存储机制：将文件切分称指定大小的数据块（默认128MB）并以多副本的方式存储在多台服务器上。 
           （默认3份）数据的切分，多副本，容错等操作对用户来说是透明的。用户操作对象还是文件。

hdfs存储概念图

   文件part-0被切分为2个数据块编号为1、3，而1、3分别有2个副本。
   文件part-1被且分为3个数据块编号为2、4、5。而3个数据块都有3个副本。
        
   replication factor：副本系数、副本因子
   All blocks in a file except the last block are the same size

HDFS架构：

        1 Master 带 N个slaves（HDFS/YARN/HBASE都是采用这种方式）
        一个文件会被拆分为多个Block：例如blocksize: 128M，则130M的文件会被拆分为两个Block：128M和2M
        namenode：Master
           职责描述: 
               1）对客户端去年供求的响应
               2）负责元数据（文本的名称，副本数、Block存放的DN）的管理
               3）对hdfs文件或者文件夹操作，比如打开，关闭，重命名等。
                 
        datanode：slaves
           职责描述：
               1）存储用户的文件对应的数据块（Block）
               2）要定期向NameNode发送心跳心跳信息，汇报借点本身及其所有的block信息，健康状况等。
               3）对block进行操作，比如对block创建，删除，以及副本的操作。

 A typical deployment has a dedicated machine that runs only the NameNode software. 
 Each of the other machines in the cluster runs one instance of the DataNode software.
 NameNode+N个DataNode

HDFS 文件读写流程：

 写文件流程：
     1、client发起请求，比如存储200M的文件（客户端根据配置文件，应该知道默认一个block的大小（一般为64M或者128M）和副本数（一般为3））。
     客户端首先将文件分为两个部分一个128M，另一个72M。
     client  divide block

    2、客户端请求发送到namenode，先存储第一个128M的文件，namenode接收到请求，。
 经过计算，namenode告诉客户端，可以存储在datanode1 ，datanode2，datanode3（顺序固定）上。

    3、客户端将存储请求发送到datanode1上，datanode1存储好后，由datanode1发送到datanode2上，datanode2存储好后，由datanode2发送到datanode3上。


    4、3个客户端存储好后都向namenode发送存储成功DONE的信息。

    5、namenode存储元数据，也就是文件的存储信息（副本数，在哪些datanode上之类的）

    6、namenode告诉客户端第一个block已经存储完成，开始存储第二个。

    7、存储完成后断开链接。

   流程总结：
   client----->namenode------>client----->datanode1--->datanode2---->datanode3----done-->namenode---->client---重复过程，知道block存储完成。


  读流程：

      1、client发起读请求。发送filename到namenode

      2、namenode返回该文件的元数据，各个block所对应的datanode

      3、client发送请求到最近的datanode上获取block信息

      4、将所有block信息拼接，输出数据。

      5、断开连接。

      client---->namenode---->client---->datanode----->client

HDFS shell：

操作：ls, get, mkdir, rm, rmdir,  put, text, cat, cp,find,  moveFromLocal, moveToLocal......
命令（command）:
    hdfs dfs -ls /
    hdfs dfs -mkdir /test
    hdfs dfs -rmdir /test
    ......

HDFS Java Api:

Java API 操作HDFS文件：
    1）IDEA + Maven 创建Java 工程（安装Maven此处不做详细说明）
    2）添加HDFS相关依赖
    3）开发Java API操作HDFS文件

     1、打开IDEA，create project，创建Maven工程

创建Maven工程

 

2、写入GroupId 和 Artifacld，点击确认下一步

 

image.png

 

3、选择安装好的Maven路径，点击确认下一步

 

选择Maven库


4、更新Maven依赖，文件内容如下：

<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 
 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>

 <groupId>com.lu.hadoop</groupId>
 <artifactId>hadoop-train1</artifactId>
 <version>1.0-SNAPSHOT</version>
 <packaging>jar</packaging>

 <name>hadoop-train1</name>
 <url>http://maven.apache.org</url>

 <properties>
   <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
   <hadoop.version>2.6.0-cdh5.7.0</hadoop.version>
 </properties>

 <repositories>
   <repository>
     <id>cloudera</id>
     <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
   </repository>
 </repositories>

 <dependencies>
   <!--添加hadoop依赖-->
   <dependency>
     <groupId>org.apache.hadoop</groupId>
     <artifactId>hadoop-client</artifactId>
     <version>${hadoop.version}</version>
   </dependency>
<!--添加单元测试的依赖-->
   <dependency>
     <groupId>junit</groupId>
     <artifactId>junit</artifactId>
     <version>4.10</version>
     <scope>test</scope>
   </dependency>
 </dependencies>
</project>

API：（在单元测试类中进行API操作）

public static final String HDFS_PATH = "hdfs://hadoop1:8020";//hdfsURI路径
FileSystem fileSystem = null;//hdfs文件系统API
Configuration configuration = null;//配置

@Before
public void setUp() throws Exception{
    configuration = new Configuration();
    fileSystem = FileSystem.get(new URI(HDFS_PATH),configuration,"root");
    System.out.println("HDFSApp:setUp");
}

/**
 * 打印
 * */
@Test
public void cat() throws IOException {

    FSDataInputStream inputStream = fileSystem.open(new Path("/hdfs_api/test/a"));
    IOUtils.copyBytes(inputStream,System.out,1024);
    inputStream.close();

}

/**
 * 创建文件夹
 * */
@Test
public void mkdir() throws IOException {
    fileSystem.mkdirs(new Path("/hdfs_api/test"));
}

/**
 * 创建文件
 * */
@Test
public void create() throws IOException {
    FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/hdfs_api/test/a"));
    fsDataOutputStream.write("aaaaa".getBytes());
    fsDataOutputStream.flush();
    fsDataOutputStream.close();
}

/**
 * 重命名
 * */
@Test
public void rename() throws IOException {
    Path oldPath = new Path("/hdfs_api/test/a");
    Path newPath = new Path("/hdfs_api/test/test1");
    fileSystem.rename(oldPath,newPath);
}

/**
 * 上传本地文件到hdfs
 * */
@Test
public void copyFromLocal() throws IOException {
    Path src = new Path("D:\\sampledb.sql");
    Path dest = new Path("/hdfs_api/test/");
    fileSystem.copyFromLocalFile(src,dest);
}

/**
 * 上传本地文件到hdfs,进度条
 * */
@Test
public void copyFromLocalWithProcess() throws IOException {
    FileInputStream fileInputStream = new FileInputStream("D:\\Download\\ideaIU-2017.3.5.tar.gz");
    InputStream in = new BufferedInputStream(fileInputStream);

    FSDataOutputStream outputStream = fileSystem.create(
            new Path("/hdfs_api/test/idea.tar.gz"), new Progressable() {
                public void progress() {
                    System.out.print(".");
                }
            });
    IOUtils.copyBytes(in,outputStream,4096);

}

@Test
public void copyToLocalFile() throws IOException {
    Path src = new Path("/hdfs_api/test/test1");
    Path dest = new Path("D:\\test1");
    fileSystem.copyToLocalFile(false,src,dest,true);
}

@Test
public void listFile() throws IOException {
    FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/hdfs_api/test/test1"));
    for(FileStatus fileStatus : fileStatuses){
        String directory = fileStatus.isDirectory() ? "DIR" : "FILE";
        short replication = fileStatus.getReplication();
        long len = fileStatus.getLen();
        String path = fileStatus.getPath().toString();
        System.out.println(directory + "\t" + replication + "\t" + len + "\t" + path);
    }

}

@After
public void tearDown() throws Exception {
    configuration = null;
    fileSystem = null;
    System.out.println("HDFSApp:tearDown");
}

--------------------------------------华丽的分割线----------------------------------------over~~~~------