zoukankan      html  css  js  c++  java
  • hadoop的dfs工具类一个【原创】

      开始没搞定插件问题,就弄了个dsf操作类,后面搞定了插件问题,这玩意也就聊胜于无了,还是丢这里算了。

      首先是一个配置,ztool.hadoop.properties

    hadoop.home.dir=G:/hadoop/hadoop-2.4.1
    hadoop.user.name=hadoop
    
    hadoop.server.ip=192.168.117.128
    hadoop.server.hdfs.port=9000

      前面两个属性后面代码会有说明的。

      属性文件的读取,方法多了,一般用commons-configuration包,我是自己把这个再整了一次,加了些自动处理,这个代码中可以无视,直接把代码中的那部分改成普通引用就好了。

      logger部分,用了logback,也是处理了一下,处理了其在linux下会莫名其妙找不到配置文件的问题。这里就不放出代码了,直接把代码中的那部分改成普通引用就好了,我就不改了。

      工具类代码如下

    package com.cnblogs.zxub.hadoop.dfs;
    
    import java.io.IOException;
    import java.net.URI;
    
    import org.apache.commons.configuration.PropertiesConfiguration;
    import org.apache.hadoop.conf.Configuration;
    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;
    import org.apache.hadoop.mapred.JobConf;
    import org.slf4j.Logger;
    
    import com.cnblogs.zxub.util.logger.ZLoggerFactory;
    import com.cnblogs.zxub.util.properties.PropertiesLoader;
    
    public class DfsUtil {
    
        private static final Logger logger = ZLoggerFactory
                .getLogger(DfsUtil.class);
    
        private final PropertiesConfiguration props = PropertiesLoader
                .getConfiguration("ztool.hadoop");
        private Configuration config = null;
        private String hdfsPath = null;
    
        private String baseDir = null;
    
        public DfsUtil(String hdfs, Configuration config) {
            // windows下设置HADOOP_HOME后,还可能找不到winutils.exe,直接自己写进去程序算了
            System.setProperty("hadoop.home.dir",
                    this.props.getString("hadoop.home.dir"));
            // 设置与dfs服务通信的用户名,省得换当前用户名,也不改配置关闭权限控制了
            System.setProperty("HADOOP_USER_NAME",
                    this.props.getString("hadoop.user.name"));
            this.hdfsPath = (hdfs == null) ? "hdfs://"
                    + this.props.getString("hadoop.server.ip") + ":"
                    + this.props.getString("hadoop.server.hdfs.port") + "/" : hdfs;
            if (config == null) {
                JobConf conf = new JobConf(DfsUtil.class);
                conf.setJobName("HdfsDAO");
                config = conf;
            }
            this.config = config;
        }
    
        public DfsUtil(Configuration conf) {
            this(null, conf);
        }
    
        public DfsUtil() {
            this(null, null);
        }
    
        public String getBaseDir() {
            return this.baseDir;
        }
    
        public void setBaseDir(String baseDir) {
            this.baseDir = baseDir;
        }
    
        public String getHdfsPath() {
            return this.hdfsPath;
        }
    
        public Configuration getConfig() {
            return this.config;
        }
    
        private String standardPath(String path) {
            if (this.baseDir == null) {
                this.baseDir = "/";
            }
            if (this.baseDir.indexOf("/") != 0) {
                this.baseDir = "/" + this.baseDir;
            }
            if (this.baseDir.lastIndexOf("/") == this.baseDir.length() - 1) {
                this.baseDir = this.baseDir.replaceFirst("/$", "");
            }
            if (path.indexOf("/") != 0) {
                path = "/" + path;
            }
            path = this.baseDir + path;
            if (path.lastIndexOf("/") == path.length() - 1) {
                path = path.replaceFirst("/$", "");
            }
            if (path.isEmpty()) {
                path = "/";
            }
            return path;
        }
    
        public void ll(String folder) throws IOException {
            folder = this.standardPath(folder);
            Path path = new Path(folder);
            FileSystem fs = FileSystem.get(URI.create(this.getHdfsPath()),
                    this.getConfig());
            FileStatus[] list = fs.listStatus(path);
            System.out.println("ll: " + folder);
            for (FileStatus f : list) {
                System.out.printf("name: %s, folder: %s, size: %d
    ", f.getPath(),
                        f.isDirectory(), f.getLen());
            }
            fs.close();
        }
    
        public void mkdirs(String folder) throws IOException {
            folder = this.standardPath(folder);
            Path path = new Path(folder);
            FileSystem fs = FileSystem.get(URI.create(this.getHdfsPath()),
                    this.getConfig());
            if (!fs.exists(path)) {
                fs.mkdirs(path);
                logger.info("create: {}.", folder);
            } else {
                logger.warn("folder [{}] already exists, mkdirs failed.", folder);
            }
            fs.close();
        }
    
        public void rm(String file) throws IOException {
            file = this.standardPath(file);
            Path path = new Path(file);
            FileSystem fs = FileSystem.get(URI.create(this.getHdfsPath()),
                    this.getConfig());
            fs.deleteOnExit(path);
            logger.info("delete: {}.", file);
            fs.close();
        }
    
        public void newFile(String file, String content) throws IOException {
            file = this.standardPath(file);
            FileSystem fs = FileSystem.get(URI.create(this.getHdfsPath()),
                    this.getConfig());
            byte[] buff = content.getBytes();
            FSDataOutputStream os = null;
            try {
                os = fs.create(new Path(file));
                os.write(buff, 0, buff.length);
                logger.info("create: {}.", file);
            } finally {
                if (os != null) {
                    os.close();
                }
            }
            fs.close();
        }
    
        public void scp(String local, String remote) throws IOException {
            remote = this.standardPath(remote);
            FileSystem fs = FileSystem.get(URI.create(this.getHdfsPath()),
                    this.getConfig());
            fs.copyFromLocalFile(new Path(local), new Path(remote));
            logger.info("copy: from [{}] to [{}]", local, remote);
            fs.close();
        }
    
        public void download(String remote, String local) throws IOException {
            remote = this.standardPath(remote);
            Path path = new Path(remote);
            FileSystem fs = FileSystem.get(URI.create(this.getHdfsPath()),
                    this.getConfig());
            fs.copyToLocalFile(path, new Path(local));
            logger.info("download: from [{}] to [{}]", remote, local);
            fs.close();
        }
    
        public void cat(String remote) throws IOException {
            remote = this.standardPath(remote);
            Path path = new Path(remote);
            FileSystem fs = FileSystem.get(URI.create(this.getHdfsPath()),
                    this.getConfig());
            FSDataInputStream fsdis = null;
            System.out.println("cat: " + remote);
            try {
                fsdis = fs.open(path);
                IOUtils.copyBytes(fsdis, System.out, 4096, false);
            } finally {
                IOUtils.closeStream(fsdis);
                fs.close();
            }
        }
    
        public static void main(String[] args) throws IOException {
            DfsUtil hdfs = new DfsUtil();
            // hdfs.setBaseDir("/test");
            // hdfs.mkdirs("/debug_in");
            // hdfs.newFile("/test.txt", "测试");
            // hdfs.rm("/test.txt");
            // hdfs.rm("/test");
            // hdfs.scp("c:/q.txt", "/");
            hdfs.ll("/");
            // hdfs.download("/test.txt", "c:/t.txt");
            // hdfs.cat("q.txt");
            // hdfs.scp("c:/din/f1.txt", "debug_in");
            // hdfs.scp("c:/din/f2.txt", "debug_in");
        }
    }
  • 相关阅读:
    windows操作系统的电脑越用越卡?简说几种原因和解决方法。
    不懂电脑也能自己重装系统,可视化傻瓜式一键重装系统不求人!!!
    023.Ubuntu常见个性化配置
    Blazor带我重玩前端(六)
    Blazor带我重玩前端(五)
    WSL2 配置及ubuntu18.04安装mysql8.0+
    用ThreadLocal来优化下代码吧
    基于canal的client-adapter数据同步必读指南
    「从零单排canal 07」 parser模块源码解析
    实用向—总结一些唯一ID生成方式
  • 原文地址:https://www.cnblogs.com/zxub/p/3884784.html
Copyright © 2011-2022 走看看