西安行业网站,在wordpress中设置mx记录,专业做企业活动的趴网站,生成链接的软件#x1f341; 博主 开着拖拉机回家带您 Go to New World.✨#x1f341; #x1f984; 个人主页——#x1f390;开着拖拉机回家_Linux,大数据运维-CSDN博客 #x1f390;✨#x1f341; #x1fa81;#x1f341; 希望本文能够给您带来一定的帮助#x1… 博主 开着拖拉机回家带您 Go to New World.✨ 个人主页——开着拖拉机回家_Linux,大数据运维-CSDN博客 ✨ 希望本文能够给您带来一定的帮助文章粗浅敬请批评指正 感谢点赞和关注 每天进步一点点加油 目录 博主 开着拖拉机回家带您 Go to New World.✨
一、FileSystem文件抽象类
1.1文件读取API
1.2文件操作API
1.3抽象FileSystem类的具体实现子类
1.4FileSystem IO输入系统相关类
1.5FileSystem IO输出系统相关类
二、HDFS的API操作
2.1测试集群版本信息
2.2文件上传下载和移动
2.3文件读写操作
2.4文件状态信息获取
2.5实战案例 一、FileSystem文件抽象类 为了提供对不同数据访问的一致接口Hadoop借鉴了Linux虚拟文件系统的概念为此Hadopo提供了一个抽象的文件系统模型FileSystemHDFS 是其中的一个实现。 FileSystem是Hadoop中所有文件系统的抽象父类它定义了文件系统所具有的基本特征和基本操作。 1.1文件读取API HadoopFileSystem操作 Java操作 Linux操作 描述 URL.openStream FileSystem.open FileSystem.create FileSystem.append URL.openStream open 打开一个文件 FSDataInputStream.read InputStream.read read 读取文件中的数据 FSDataInputStream.write OutputStream.write write 向文件中写入数据 FSDataInputStream.close FSDataOutputStream.close InputStream.close OutputStream.close close 关闭一个文件 FSDataInputStream.seek RandomAccessFile.seek lseek 改变文件读写位置 FileSystem.getContentSummary du/wc 获取文件存储信息 1.2文件操作API HadoopFileSystem操作 Java操作 Linux操作 描述 FileSystem.getFileStatus FileSystem.get* File.get* stat 获取文件/目录的属性 FileSystem.set* File.set* chomd 修改文件属性 FileSystem.createNewFile File.createNewFile create 创建一个文件 FileSystem.delete File.delete remove 删除一个文件 FileSystem.rename File.renameTo rename 移动或先修改文件/目录名 FileSystem.mkdirs File.mkdir mkdir 创建目录 FileSystem.delete File.delete rmdir 从一个目录下删除一个子目录 FileSystem.listStatus File.list readdir 读取一个目录下的项目 FileSystem.setWorkingDirectory getcwd/getwd 返回当前工作目录 FileSystem.setWorkingDirectory chdir 更改当前的工作目录
1.3抽象FileSystem类的具体实现子类 1.4FileSystem IO输入系统相关类 1.5FileSystem IO输出系统相关类 二、HDFS的API操作 2.1测试集群版本信息 2.2文件上传下载和移动
/*** 本地文件上传到 HDFS** param srcPath 本地路径 文件名* param dstPath Hadoop路径* param fileName 文件名*/
def copyToHDFS(srcPath: String, dstPath: String, fileName: String): Boolean {var path new Path(dstPath)val fileSystem: FileSystem path.getFileSystem(conf)val isFile new File(srcPath).isFile// 判断路径是否存在val existDstPath: Boolean fileSystem.exists(path)if (!existDstPath) {fileSystem.mkdirs(path)}// 本地文件存在if (isFile) {// HDFS 采用 路径 文件名path new Path(dstPath File.separator fileName)// false: 是否删除 目标文件,false: 不覆盖fileSystem.copyFromLocalFile(false, false, new Path(srcPath), path)return true}false
}/*** Hadoop文件下载到本地** param srcPath hadoop 源文件* param dstPath 目标文件* param fs 文件访问对象*/
def downLoadFromHDFS(srcPath: String, dstPath: String, fs: FileSystem): Unit {val srcPathHDFS new Path(srcPath)val dstPathLocal new Path(dstPath)// false: 不删除源文件fs.copyToLocalFile(false, srcPathHDFS, dstPathLocal)
}/*** 检查Hadoop文件是否存在并删除** param path HDFS文件*/
def checkFileAndDelete(path: String, fs: FileSystem) {val dstPath: Path new Path(path)if (fs.exists(dstPath)) {// false: 是否递归删除否fs.delete(dstPath, false)}
}/*** 获取指定目录下正则匹配后的文件列表** param dirPath hdfs路径* param regexRule 正则表达式 ,如^(?!.*[.]tmp$).*$ ,匹配非 .tmp结尾的文件*/def listStatusHDFS(dirPath: String, regexRule: String, fs: FileSystem): util.ArrayList[Path] {val path new Path(dirPath)val pattern: Pattern Pattern.compile(regexRule)// 匹配的文件val fileList new util.ArrayList[Path]()val fileStatusArray: Array[FileStatus] fs.listStatus(path)for (fileStatus - fileStatusArray) {// 文件 全路径val filePath: Path fileStatus.getPath()val fileName: String filePath.getName.toLowerCaseif (regexRule.equals()) {// 如果匹配规则为空 则获取目录下的全部文件fileList.add(filePath)log.info(match file : fileName)} else {// 正则匹配文件if (pattern.matcher(fileName).matches()) {fileList.add(filePath)log.info(match file : fileName)}}}fileList
}/*** 文件移动或重命名到指定目录, 如文件00000 重命名为00001** param srcPath 源文件路径* param dstPath 源文件路径* param fs 文件操作对象*/
def renameToHDFS(srcPath: String, dstPath: String, fs: FileSystem): Boolean {var renameFlag falseval targetPath new Path(dstPath)// 目标文件存在先删除if (fs.exists(targetPath)) {fs.delete(targetPath, false)}renameFlag fs.rename(new Path(srcPath), targetPath)if (renameFlag) {log.info(renamed file srcPath to targetPath success!)} else {log.info(renamed file srcPath to targetPath failed!)}renameFlag
}
2.3文件读写操作 Hadoop抽象文件系统也是使用流机制进行文件的读写。Hadoop抽象文件系统中用于读文件数据的流是FSDataInputStream对应地写文件通过抽象类FSDataOutputStream实现。 /*** 读取HDFS文件** param inPutFilePath 源文件路径* param fs 文件操作对象*/
def readFromHDFS(inPutFilePath: String, OutputFilePath: String, fs: FileSystem) {var fSDataInputStream: FSDataInputStream nullvar bufferedReader: BufferedReader nullval srcPath new Path(inPutFilePath)if (fs.exists(srcPath)) {val fileStatuses: Array[FileStatus] fs.listStatus(srcPath)for (fileStatus - fileStatuses) {val filePath: Path fileStatus.getPath// 判断文件大小if (fs.getContentSummary(filePath).getLength 0) {fSDataInputStream fs.open(filePath)bufferedReader new BufferedReader(new InputStreamReader(fSDataInputStream))var line bufferedReader.readLine()while (line ! null) {print(line \n) // 打印line bufferedReader.readLine()}}}}fSDataInputStream.close()bufferedReader.close()
}/*** 读取HDFS文件, 处理完成 重新写入** param inPutFilePath 源文件路径* param OutputFilePath 输出文件到新路径* param fs 文件操作对象*/
def writeToHDFS(inPutFilePath: String, OutputFilePath: String, fs: FileSystem) {var fSDataInputStream: FSDataInputStream nullvar fSDataOutputStream: FSDataOutputStream nullvar bufferedReader: BufferedReader nullvar bufferedWriter: BufferedWriter nullval srcPath new Path(inPutFilePath)var count 0if (fs.exists(srcPath)) {val fileStatuses: Array[FileStatus] fs.listStatus(srcPath)for (fileStatus - fileStatuses) {val filePath: Path fileStatus.getPath// 判断文件大小if (fs.getContentSummary(filePath).getLength 0) {fSDataInputStream fs.open(filePath)bufferedReader new BufferedReader(new InputStreamReader(fSDataInputStream))val outputFilePath new Path(OutputFilePath count)fSDataOutputStream fs.create(outputFilePath)bufferedWriter new BufferedWriter(new OutputStreamWriter(fSDataOutputStream, UTF-8))var line bufferedReader.readLine()while (line ! null) {val bytes: Array[Byte] line.getBytes(UTF-8)bufferedWriter.write(new String(bytes) \n)line bufferedReader.readLine()}bufferedWriter.flush()count 1}}}fSDataInputStream.close()bufferedReader.close()bufferedWriter.close()
}
测试结果如下 2.4文件状态信息获取 FileSystem. getContentSummary提供了类似Linux命令du、df提供的功能。du表示disk usage它会报告特定的文件和每个子目录所使用的磁盘空间大小命令df则是diskfree的缩写用于显示文件系统上已用的和可用的磁盘空间的大小。du、df是Linux中查看磁盘和文件系统状态的重要工具。 getContentSummary方法的输入是一个文件或目录的路径输出是该文件或目录的一些存储空间信息这些信息定义在ContentSummary包括文件大小、文件数、目录数、文件配额已使用空间和已使用文件配额等。 /*** HDFS路径下文件信息统计** param dirPath hdfs路径**/def listHDFSStatus(dirPath: String, fs: FileSystem) {val path new Path(dirPath)// 匹配的文件val contentSummary: ContentSummary fs.getContentSummary(path)println(/tmp/kangll 目录下子目录个数: , contentSummary.getDirectoryCount)println(/tmp/kangll 目录下文件个数: , contentSummary.getFileCount)println(/tmp/kangll 目录下文件大小 , contentSummary.getLength)println(/tmp/kangll 目录下文件和子目录个数: , contentSummary.getFileAndDirectoryCount)}/tmp/kangll目录信息获取结果 2.5实战案例 案例说明 HDFS 文件清理 根据文件大小、个数、程序休眠时间控制 匀速 批量删除 HDFS 文件当文件越大 需要配置 删除个数更少休眠时间更长防止 NameNode 负载过大减轻DataNode磁盘读写压力从而不影响线上业务情况下清理过期数据。 package com.kangll.common.utilsimport java.text.SimpleDateFormat
import java.util.concurrent.TimeUnit
import java.util.{Calendar, Date, Properties}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{ContentSummary, FileStatus, FileSystem, Path}
import org.apache.log4j.Loggerimport scala.collection.mutable.ListBuffer/** ***************************************************************************************** auther kangll * date 2023/09/12 12:10 * desc HDFS 文件清理 根据文件大小、个数、程序休眠时间控制 匀速 批量删除* HDFS 文件当文件越大 需要配置 删除个数更少休眠时间更长防止* NameNode 负载过大减轻DataNode磁盘读写压力从而不影响线上业务下删除*** 1.遍历文件夹下的文件个数据 当遍历的文件夹下的文件个数到达阈值时 将* 文件所述的 父路径直接删除** ****************************************************************************************/
object CleanHDFSFileUtil {// 删除文件总数统计var HDFS_FILE_SUM 0// 批次删除文件个数显示var HDFS_FILE_BATCH_DEL_NUM 0val start System.currentTimeMillis()/**** param fs 文件操作对象* param pathName 文件根路径* param fileList 批次清理的 buffer* param saveDay 根据文件属性 获取文件创建时间 选择文件保留最近的天数* param sleepTime 休眠时间防止一次性删除太多文件 导致 datanode 文件负载太大* param fileBatchCount 批次删除文件的个数 相当于是 上报到 namenode 文件清理队列的大小参数越大 队列越大datanode 磁盘负载相对来说就高* return*/def listPath(fs: FileSystem, pathName: String, fileList: ListBuffer[String], saveDay: Int, sleepTime: Long, fileBatchCount: Int): ListBuffer[String] {val fm new SimpleDateFormat(yyyy-MM-dd)// 获取当前时间val currentDay fm.format(new Date())val dnow fm.parse(currentDay)val call Calendar.getInstance()call.setTime(dnow)call.add(Calendar.DATE, -saveDay)// 获取保留天前的时期val saveDayDate call.getTime// 遍历文件val fileStatuses fs.listStatus(new Path(pathName))for (status - fileStatuses) {// 获取到文件名val filePath status.getPathif (status.isFile) {// 获取到文件修改时间val time: Long status.getModificationTimeval hdfsFileDate fm.parse(fm.format(new Date(time)))if (saveDayDate.after(hdfsFileDate)) {fileList filePath.toString// 获取文件个数val cs: ContentSummary fs.getContentSummary(filePath)HDFS_FILE_SUM cs.getFileCount.toIntHDFS_FILE_BATCH_DEL_NUM cs.getFileCount.toIntif (HDFS_FILE_BATCH_DEL_NUM fileBatchCount) {val end System.currentTimeMillis()println()println( 遍历文件数量达到 HDFS_FILE_BATCH_DEL_NUM 个删除HDFS文件 )println( 休眠 sleepTime S )println( 删除文件总数 HDFS_FILE_SUM )println( 程序运行时间 (end - start) / 1000 s )println()HDFS_FILE_BATCH_DEL_NUM 0TimeUnit.MILLISECONDS.sleep(sleepTime)}// 文件删除根据绝对路径删除println( 删除文件 filePath )// 递归删除fs.delete(filePath, true)}} else {// 递归文件夹listPath(fs, filePath.toString, fileList, saveDay, sleepTime, fileBatchCount)}}println( 删除文件总数 HDFS_FILE_SUM )fileList}/*** 删除空文件夹** param fs 文件操作对象* param pathName 路径* param pathSplitLength 文件按照/拆分后的长度*/def delEmptyDirectory(fs: FileSystem, pathName: String, pathSplitLength: Int) {// 遍历文件val fileStatuses fs.listStatus(new Path(pathName))for (status - fileStatuses) {if (status.isDirectory) {val path: Path status.getPath// /kangll/winhadoop/temp/wmall_batch_inout/day/1660878372 7val delPathSplitLength path.toString.substring(6, path.toString.length).split(/).length// filePath /kangll/winhadoop/temp/wmall_batch_inout/day 子时间戳文件夹两个// val hdfsPathListCount fileStatuses.lengthval hdfsPathListCount fs.listStatus(path).lengthif (delPathSplitLength pathSplitLength hdfsPathListCount 0) {println( 删除空文件夹 path )fs.delete(path, true)}}}}def main(args: Array[String]): Unit {val logger Logger.getLogger(CleanHDFSFileUtil)val conf new Configuration()conf.set(fs.hdfs.impl, org.apache.hadoop.hdfs.DistributedFileSystem)conf.set(fs.file.impl, org.apache.hadoop.fs.LocalFileSystem)val fs FileSystem.get(conf)val fileList new ListBuffer[String]val hdfsDir if (args.size 0) args(0).toString else System.exit(0).toStringval saveDay if (args.size 1) args(1).toInt else 2val sleepTime if (args.size 2) args(2).toLong else 10val fileBatchCount if (args.size 3) args(3).toInt else 5/*默认不启用文件夹删除参数为 文件夹绝对路径Split后的数组长度如 路径 /winhadoop/temp/wmall_batch_inout/thirty 配置为 7*/val pathSplitLength if (args.size 4) args(4).toInt else 20// 删除文件listPath(fs, hdfsDir, fileList, saveDay, sleepTime, fileBatchCount)// 删除空文件夹delEmptyDirectory(fs, hdfsDir, pathSplitLength)fs.close()}
}调用脚本
#
# 脚本功能: 过期文件清理
# 作 者: kangll
# 创建时间: 2023-09-14
# 修改内容: 控制删除文件的批次个数程序休眠时间传入
# 当前版本: 1.0v
# 调度周期: 一天一次
# 脚本参数: 删除文件夹、文件保留天数、程序休眠时间、批次删除个数
# 1.文件根路径子文件夹递归遍历
# 2.文件保留天数
# 3.程序休眠时间 防止 DataNode 删除文件负载过大单位 秒
# 4.批次删除文件个数 ,如配置 100当满足文件个数100时 整批执行 delete,紧接着程序休眠
# 5.默认不启用文件夹删除也就是不传参参数为 文件夹绝对路径Split后的数组长度
# /winhadoop/temp/wmall_batch_inout/thirty/时间戳/ Split后 长度为7默认删除时间戳文件夹
#### 对应的新删除程序
jarPath/hadoop/project/del_spark2-1.0-SNAPSHOT.jar### 集群日志
java -classpath $jarPath com.kangll.common.utils.CleanHDFSFileUtil /spark2-history 3 10 100 参考 :
hadoop抽象文件系统filesystem框架介绍_org.apache.hadoop.fs.filesystem_souy_c的博客-CSDN博客
Hadoop FileSystem文件系统的概要学习 - 回眸,境界 - 博客园
hadoop抽象文件系统filesystem框架介绍_org.apache.hadoop.fs.filesystem_souy_c的博客-CSDN博客
