Hadoop Java API操作及读取序列化文件（04-05-06）

news2024/11/15 17:20:55

针对于04-05-06班级整合。

1.创建java项目

2.修改pom.xml文件

添加依赖

<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.1.4</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.30</version>
    </dependency>
</dependencies>

3.创建一个名为HdfsTest类，在类中编写单元测试代码测试API中的每个方法

（1）查看HDFS根目录下的文件或子目录，并将名称打印在控制台

//单元测试  每个操作放在一个测试方法中，  选中方法名---》右击  运行testList()
@Test
public void testList() throws Exception{

    //1.创建配置对象
    Configuration conf = new Configuration();
    //2.获取文件系统对象  yt启动集群的用户名
    // FileSystem fs = FileSystem.get(conf);
    FileSystem fs = FileSystem.get(new URI("hdfs://master:8020") ,conf,"yt");
    //3.获取HDFS根目录下的子目录和文件
    FileStatus  []files =fs.listStatus( new Path("/"));
    //4.循环所有的files
    for(FileStatus f:files){
        //判断是文件还是目录
        if(f.isFile()){
            System.out.println(  "f:  "+f.getPath().getName() );
        }else {
            System.out.println(" d:    "+f.getPath().getName());
        }
    }

    //关闭资源
    fs.close();

}

控制台输出的结果：

（2）创建目录

//创建目录
@Test
public void testMkdirs() throws Exception{

    //1.创建配置对象
    Configuration conf = new Configuration();
    //2.获取文件系统对象  yt启动集群的用户名
    FileSystem fs = FileSystem.get(new URI("hdfs://master:8020") ,conf,"yt");

    fs.mkdirs(new Path("/hdfs/2306"));


    //关闭资源
    fs.close();

}

测试结果：

（3）删除非空目录

//删除文件或目录
@Test
public void testDelete() throws Exception{

    //1.创建配置对象
    Configuration conf = new Configuration();
    //2.获取文件系统对象  yt启动集群的用户名
    FileSystem fs = FileSystem.get(new URI("hdfs://master:8020") ,conf,"yt");

    fs.delete(new Path("/output-2302-order"),true);


    //关闭资源
    fs.close();

}

（4）文件上传

//文件上传
@Test
public void testCopyFromLocalFile() throws Exception{

    //1.创建配置对象
    Configuration conf = new Configuration();
    //2.获取文件系统对象  yt启动集群的用户名
    FileSystem fs = FileSystem.get(new URI("hdfs://master:8020") ,conf,"yt");

    fs.copyFromLocalFile(new Path("D:\\software\\apache-maven-3.8.6.rar"),new Path("/hdfs/2306/"));
    //关闭资源
    fs.close();

}

（5）文件下载

//文件下载
@Test
public void testCopyToLocalFile() throws Exception{

    //1.创建配置对象
    Configuration conf = new Configuration();
    //2.获取文件系统对象  yt启动集群的用户名
    FileSystem fs = FileSystem.get(new URI("hdfs://master:8020") ,conf,"yt");

    fs.copyToLocalFile(new Path("/bigdata/order.txt"),new Path("D:\\software\\"));
    //关闭资源
    fs.close();

}

（6）读取HDFS上的序列化文件，将数据保存到D:\\software\\raceData.txt

//读取序列化文件中的数据保存到本次磁盘
@Test
public void testSequenceFileRead() throws Exception{

    //1.创建配置对象
    Configuration conf = new Configuration();
    //2.获取文件系统对象  yt启动集群的用户名
    FileSystem fs = FileSystem.get(new URI("hdfs://master:8020") ,conf,"yt");

    SequenceFile.Reader reader = new SequenceFile.Reader(fs,new Path("/ouput-2301-select/part-m-00000"),conf);
    //创建文件对象
    File file = new File("D:\\software\\raceData.txt");
    //创建字节流对象
    OutputStream  out = new FileOutputStream(file);
    //创建缓冲字符输出流对象
    BufferedWriter  bw =new BufferedWriter( new OutputStreamWriter(out) );
    Text key =new Text();
    Text val =new Text();
    while(  reader.next(key,val)   ){
        bw.write(  key.toString()+"\t" +val.toString()  );
        bw.newLine();
    }
    bw.flush();
    //关闭资源
    bw.close();
    out.close();
    fs.close();

}