本篇讲解如何通过Eclipse 编写代码去操作分析hdfs 上的文件。
1、在eclipse 下新建Map/Reduce Project项目。如图： 
项目建好后，会默认加载一系列相应的jar包。
下面还有很多包。
2、我们新建Java 类就可以了。代码如下：
package org.hadoop.examples;
import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
public class operateHdfs {
public static void main(String args[]) {
// 测试 创建新文件
String contents = "hello world 张三\n--created by lyf\n";
// 创建文件夹
// creatFolder("/input/lyf2/");
// 创建文件
// createFile("/input/demo.txt", contents);
// 删除文件(文件和文件夹)
// deleteFile("/input/first.txt");
// deleteFile("/input/lyf22");
// 上传文件到指定位置
// uploadFile("E:/myTest.txt", "/input");
// 重命名(文件和文件夹--文件夹下不含文件)
// renameFile(new Path("/input/lyf"),new Path("/input/test"));
// 获取指定路径下的所有文件
// getUnderDirAllFile("/input/lyf");
// 判断文件是否存在
// findFileIsExit(new Path("/input/lyf/first.txt"));
// 查看某个文件在HDFS集群的位置
// findLocationOnHadoop(new Path("/input/lyf/demo.txt"));
// 文件内容追加
// appendFile("/input/demo.txt", "/input/demo.txt");
// 读取指定文件
// readFile("/input/demo.txt");
// 写入文件 和创建文件一样
// writeFile("/input/demo.txt");
// 下载文件
// downloadFile(new Path("/input/demo.txt"),new Path("D:/demo222.txt"));
}
// 创建文件目录
public static void creatFolder(String folderPath) {
// master
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
Path demoDir = new Path(folderPath);
try {
FileSystem fs = FileSystem.get(conf);
if (fs.mkdirs(demoDir)) {
System.out.println("文件夹创建成功！");
} else {
System.out.println("文件夹创建失败！");
}
} catch (IOException e) {
System.out.println("出现异常！");
}
}
// 创建新文件（直接生成指定路径下的first.txt，即：/eclipse/first.txt）
public static void createFile(String dst, String contents) {
// master
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
System.out.println("-----------:" + conf);
try {
FileSystem fs = FileSystem.get(conf);
Path dstPath = new Path(dst); // 目标路径
// 打开一个输出流
FSDataOutputStream outputStream = fs.create(dstPath);
outputStream.write(contents.getBytes());
outputStream.writeUTF(contents);
outputStream.close();
fs.close();
System.out.println("文件创建成功！");
} catch (IOException e) {
e.printStackTrace();
System.out.println("出现异常----文件创建失败！");
}
}
// 删除指定文件
public static void deleteFile(String file) {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
FileSystem fs;
try {
fs = FileSystem.get(conf);
Path path = new Path(file);
fs.delete(path);
fs.close();
System.out.println("文件删除成功" + conf);
} catch (IOException e) {
e.printStackTrace();
System.out.println("出现异常---文件删除失败" + conf);
}
}
// 上传指定文件到指定路径
public static void uploadFile(String s, String d) {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
FileSystem fs;
try {
fs = FileSystem.get(conf);
Path src = new Path(s);
Path dst = new Path(d);
fs.copyFromLocalFile(src, dst);
fs.close();
System.out.print("文件上传成功！" + conf);
} catch (IOException e) {
e.printStackTrace();
System.out.print("异常---文件上传失败！" + conf);
}
}
// 获取指定路径下的所有文件
public static void getUnderDirAllFile(String rootPath) {
Path targetDir = new Path(rootPath);
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "192.168.1.216:9000");
try {
FileSystem fs = FileSystem.get(conf);
FileStatus[] fileStatus = fs.listStatus(targetDir);
for (FileStatus file : fileStatus) {
System.out.println(file.getPath() + "--" + file.getGroup() + "--" + file.getBlockSize() + "--"
+ file.getLen() + "--" + file.getModificationTime() + "--" + file.getOwner());
}
} catch (IOException e) {
e.printStackTrace();
System.out.println("异常----文件获取失败！");
}
}
// 判断文件是否存在
public static void findFileIsExit(Path filePath) {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
try {
FileSystem fs = FileSystem.get(conf);
if (fs.exists(filePath)) {
System.out.println("文件存在！");
} else {
System.out.println("文件不存在！");
}
} catch (IOException e) {
e.printStackTrace();
System.out.println("出现异常！");
}
}
// 读取指定路径的文件
public static void readFile(String path) {
Configuration conf = new Configuration();
//conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
try {
FileSystem fs = FileSystem.get(URI.create("hdfs://192.168.1.216:9000"),conf);
FSDataInputStream ins = fs.open(new Path(path));
// readUTF 读取的文件必须是writeUTF 写入的，否则报 EOFException
String content = ins.readUTF();
System.out.println("文件内容" + content);
ins.close();
// fs.close();
} catch (IOException e) {
e.printStackTrace();
System.out.println("读取内容失败！");
}
}
// 写如文件
public static void writeFile(String path) {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
try {
FileSystem fs = FileSystem.get(conf);
FSDataOutputStream ous = fs.create(new Path(path));
ous.writeUTF("欢迎写入文件！");
ous.close();
System.out.println("文件写入成功！");
} catch (IOException e) {
e.printStackTrace();
System.out.println("文件写入失败！");
}
}
// 追加内容到
public static void appendFile(String filePath, String addPath) {
Configuration conf = new Configuration();
conf.set("fs.defaultSF", "hdfs://192.168.1.216:9000");
conf.setBoolean("dfs.support.append", true);
try {
FileSystem fs = FileSystem.get(URI.create(filePath), conf);
// 要追加的文件流，inpath为文件
InputStream in = new BufferedInputStream(new FileInputStream(filePath));
OutputStream out = fs.append(new Path(addPath));
IOUtils.copyBytes(in, out, 4096, true);
System.out.println(" 文件追加成功！" + conf);
} catch (Exception e) {
e.printStackTrace();
System.out.println("出现异常---文件追加失败！" + conf);
}
}
// 文件重命名(指定文件改成指定名称)
public static void renameFile(Path oldName, Path newName) {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://192.168.1.216:9000");
try {
FileSystem fs = FileSystem.get(conf);
if (fs.rename(oldName, newName)) {
System.out.println("文件重命名---成功！");
} else {
System.out.println("文件重命名---失败！");
}
} catch (IOException e) {
e.printStackTrace();
System.out.println("异常----文件重命名---失败！");
}
}
// 查看某个文件在HDFS集群的位置
public static void findLocationOnHadoop(Path rootPath) {
Configuration conf = new Configuration();
// conf.set("fs.defultFS", "hdfs://192.168.1.216:9000");
try {
FileSystem fs = FileSystem.get(URI.create("hdfs://192.168.1.216:9000"), conf);
FileStatus fileStaus = fs.getFileStatus(rootPath);
BlockLocation[] bloLocations = fs.getFileBlockLocations(fileStaus, 0, fileStaus.getLen());
for (int i = 0; i < bloLocations.length; i++) {
System.out.println("block_" + i + "_location:" + bloLocations[i].getHosts()[0]);
}
System.out.println("获取文件在HDFS集群位置成功！");
} catch (IOException e) {
e.printStackTrace();
System.out.println("异常 ----获取文件在HDFS集群位置失败！");
}
}
// 下载文件到指定位置
public static void downloadFile(Path filePath,Path downPath){
Configuration conf = new Configuration();
try {
FileSystem fs = FileSystem.get(URI.create("hdfs://192.168.1.216:9000"),conf);
fs.copyToLocalFile(filePath, downPath);
System.out.println("文件下载成功！");
} catch (IOException e) {
e.printStackTrace();
System.out.println("----异常--- 文件下载失败！");
}
}
}
其中hdfs://192.168.1.216 是HDFS文件服务器的地址，你改成你文件服务地址就好了。
这时候是已经可以运行了，但控制台会显示如下错误：
log4j:WARN No appenders could be found for logger (org.apache.Hadoop.metrics2.lib.MutableMetricsFactory).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
解决方式：在src 文件下新建 log4j.properties 文件里面的内容如下：
log4j.rootLogger=INFO, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n log4j.appender.logfile=org.apache.log4j.FileAppender log4j.appender.logfile.File=target/spring.log log4j.appender.logfile.layout=org.apache.log4j.PatternLayout log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
至此文件操作就已经完成了。
时间： 2024-07-30 13:38:54