HDFS Java api操作-cnblog

HDFS Java API操作

1 启动服务

zkServer.sh start (每个节点都要启动)
#下面的只在主节点上启动就行
start-all.sh
#查看
jps

启动后如图所示：

在浏览器访问Hadoop，点击Browse the file system 查看HDFS文件系统的目录

2 创建Maven项目

file -> new -> project

3 导入依赖

引入hadoop-common、hadoop-hdfs、hadoop-client

<dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.3.6</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>2.7.4</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>3.3.6</version></dependency>

4 新建文件夹

创建新的Java包命名为jqe

5 创建新的class文件

命名为Testfile

6 创建目录

在Testfiles类中添加一个用于创建文件目录的createDir()测试方法，该方法可以判断创建的目录是否存在，如果不存在，就创建一个新的目录，代码如下：

package jqe;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.Test;
import java.net.URI;public class Testfile {@Testpublic void createDir() throws Exception{Configuration conf = new Configuration();//置顶HDFS集群地址URI uri = new URI("hdfs://ns");//创建文件系统对象FileSystem fs = FileSystem.get(uri, conf, "root");//创建文件目录Path dirs = new Path("/jqe/t1");try{//判断文件目录是否存在，若不存在就创建，若存在则不创建boolean re = fs.exists(dirs);if(!re){fs.mkdirs(dirs);System.out.println("文件目录已创建");}else{System.out.println("文件目录已存在");}}catch(Exception e){System.err.println("创建文件目录错误");e.printStackTrace();}finally{fs.close();}}}

运行createDir()测试方法,运行成功如图所示：

如果出现了java.net.UnknownHostException: ns，即本地无法解析 ns 这个主机名时失败，无法找到识别它对应的 IP 地址。这通常是由于本地（运行 Java 代码的 Windows 机器）无法解析 HDFS 集群的命名服务 ns 导致的。

解决方法：

1. 确认Hadoop集群中ns的配置

# 在集群节点上执行，查看命名服务配置
grep "fs.defaultFS" /usr/local/hadoop/etc/hadoop/core-site.xml
grep "dfs.nameservices" /usr/local/hadoop/etc/hadoop/hdfs-site.xml
grep "dfs.ha.namenodes.ns" /usr/local/hadoop/etc/hadoop/hdfs-site.xml

2. 在本地Windows机器上配置hosts文件

找到Windows的hosts文件，路径C:\Windows\System32\drivers\etc\hosts，添加集群节点的IP和主机名映射

# Hadoop 集群节点 IP 和主机名（示例）
192.168.204.131  hadoop01  # nn1 的 IP 和主机名
192.168.204.132  hadoop02  # nn2 的 IP 和主机名
192.168.204.131  ns        # 关键：将 ns 映射到其中一个 namenode 的 IP（或负载均衡 IP）

3. 验证本地是否能解析ns

ping ns

4. 补充Hadoop配置文件到Java项目

为避免代码中因缺少集群配置导致的其他问题，将 Hadoop 集群的 core-site.xml 和 hdfs-site.xml 复制到 Java 项目的 src/main/resources 目录下。

5. 在HDFS中为windos用户授权

登录 Hadoop 集群的 namenode 节点（如 hadoop01），通过 hdfs dfs 命令为 Windows 用户赋予 HDFS 根目录的写入权限：

# 切换到 root 超级用户（确保集群以 hdfs 用户启动）
su - root# 为 HDFS 根目录添加其他用户的写入权限（临时测试用）
hdfs dfs -chmod o+w /# 或更规范的方式：创建一个属于 Vfrti 的目录，并赋予权限(Vfrti是我的Windows用户名)
hdfs dfs -mkdir /Vfrti
hdfs dfs -chown Vfrti:supergroup /Vfrti
hdfs dfs -chmod 755 /user/Vfrti

7 上传本地文件

在Testfiles类中添加一个用于上传文件的putFiles()测试方法，该方法用于从Windows系统本地上传多个文件到群集，代码如下：

package jqe;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.conf.Configuration;
import org.junit.Test;
import java.net.URI;public class Testfile {@Testpublic void putFiles() throws Exception{FileSystem fs = FileSystem.get(new URI("hdfs://ns"),new Configuration());//定义本地文件路径（两个路径位置）Path win_local1 = new Path("D:/data1.txt");Path win_local2 = new Path("D:/data2.txt");//定义HDFS存储位置Path dfs = new Path("/jqe/t1");//上传多路径文件fs.copyFromLocalFile(win_local1,dfs);fs.copyFromLocalFile(win_local2,dfs);//文件存放数组FileStatus files[] = fs.listStatus(dfs);for (FileStatus file:files){System.out.println(file.getPath());}fs.close();}
}

8 下载文件到本地

载TestFiles类种添加一个用于下载文件的getFiles()测试方法，该方法通过正则表达式过滤出以“txt”为文件扩展名的文件并下载，代码如下：

package jqe;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.conf.Configuration;
import org.junit.Test;
import java.net.URI;public class Testfile {@Testpublic void getFiles() throws Exception{System.setProperty("hadoop.home.dir", "D:/hadoop3.2.2");FileSystem fs = FileSystem.get(new URI("hdfs://ns"),new Configuration());//定义要将文件保存到的本地路径Path dispath = new Path("D:/data");//定义要下载HDFS文件的存储位置Path dfs = new Path("/input");//递归列出该目录下的所有文件（不包括文件夹，布尔值表示是否递归）RemoteIterator<LocatedFileStatus>listFiles=fs.listFiles(dfs,true);while(listFiles.hasNext()){//得到下一个文件并pop出listFilesLocatedFileStatus next = listFiles.next();//打印文件路径System.out.println(next.getPath());//过滤文件，将以“.txt”为扩展名的文件下载到本地MyPathFilter myPathFilter = new MyPathFilter(".*\\.txt");if (!myPathFilter.accept(next.getPath())){//保存HDFS文件到本地fs.copyToLocalFile(next.getPath(),dispath);System.out.println("下载的文件为"+next.getPath().getName());}}//关闭连接fs.close();}/** 实现PathFilter接口，自定义文件过滤类* @author jqe* */class MyPathFilter implements PathFilter{String reg = null;public MyPathFilter(String reg){this.reg = reg;}@Overridepublic boolean accept(Path path) {if (!path.toString().matches(reg)){return true;}return false;}}
}

结果如图所示：

注：

如果在 Windows 系统中运行 Hadoop 客户端代码时，未设置 HADOOP_HOME 环境变量，且缺少 Windows 平台所需的 winutils.exe 工具，会导致文件操作失败。

System.setProperty("hadoop.home.dir", "D:/hadoop3.2.2");这一行代码指定了winutils.exe的位置

winutils.exe下载地址：https://github.com/cdarlint/winutils下载与虚拟机对应的hadoop版本

在宿主机里面新建一个文件夹命名为Hadoopx.x.x(对应版本号)，在里面再创建一个子文件夹命名为bin，把winutils.exe放进去，根据自己的路径修改代码即可。

9 删除文件

package jqe;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.conf.Configuration;
import org.junit.Test;
import java.net.URI;public class Testfile {@Testpublic void deleteFiles() throws Exception{FileSystem fs = FileSystem.get(new URI("hdfs://ns"),new Configuration(),"root");//删除文件，第2个参数为是否递归删除文件夹及文件夹下的数据文件fs.delete(new Path("/input/data20251022.txt"),true);//关闭连接fs.close();}
}

输出结果如下（不会有任何返回，可以通过查看HDFS集群/input目录下的data20251022.txt文件是否被删除）：

10 写入数据

package jqe;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.conf.Configuration;
import org.junit.Test;
import java.net.URI;public class Testfile {@Testpublic void writeHDFS() throws Exception{FileSystem fs = FileSystem.get(new URI("hdfs://ns"),new Configuration(),"root");try{//定义文件名Path dfs = new Path("/jqe/newfile.txt");//创建输出流对象FSDataOutputStream create = null;//若文件不存在，就创建文件并写入数据；若文件存在，就追加数据if(!fs.exists(dfs)){//创建新的文件，“false”表示不覆盖原文件create = fs.create(dfs,false);//写入数据create.writeBytes("This is a HDFS file!\n");create.writeBytes("Welcome to Hadoop!\n");System.out.println("新的数据写入成功");}else{//文件存在，载文件种追加新的数据create = fs.append(dfs);create.writeBytes("Do you know HDFS?\n");System.out.println("新的数据追加成功");}}catch (Exception e){System.err.println("写入数据错误");e.printStackTrace();}finally {//关闭连接fs.close();}}
}

运行成功如图所示：

基于这个文件已经存在的情况下再运行一次会出现新的一行：

11 读取数据

package jqe;import org.apache.hadoop.fs.*;
import org.apache.hadoop.conf.Configuration;
import org.junit.Test;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.URI;public class Testfile {@Testpublic void readHDFS() throws Exception{FileSystem fs = FileSystem.get(new URI("hdfs://ns"),new Configuration(),"root");//读取的HDFS文件路径Path path = new Path("hdfs://ns/jqe/newfile.txt");if (fs.exists(path)){System.out.println("Exists!");try{//此为Hadoop读取数据类型FSDataInputStream is = fs.open(path);//创建InputStreamReader对象InputStreamReader inputStreamReader = new InputStreamReader(is,"UTF-8");String line = null;//将数据放入缓冲区BufferedReader reader = new BufferedReader(inputStreamReader);//从缓冲区中读取数据int i = 0;while ((line = reader.readLine()) != null){i++;//打印每行的数据System.out.println("line"+i+"="+line);}}catch (Exception e){System.out.println(e);}}else{System.out.println("不存在");}}

运行成功如下图：