经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Hadoop » 查看文章
Hadoop学习(2)-java客户端操作hdfs及secondarynode作用
来源：cnblogs　　作者：两千个秘密　　时间：2019/7/24 8:36:47　　对本文有异议

首先要在windows下解压一个windows版本的hadoop

然后在配置他的环境变量，同时要把hadoop的share目录下的hadoop下的相关jar包拷贝到esclipe

然后Build Path

下面上代码

  1. import java.io.BufferedReader;
  2. import java.io.FileInputStream;
  3. import java.io.IOException;
  4. import java.io.InputStreamReader;
  5. import java.net.URI;
  6. import java.net.URISyntaxException;
  7. import java.util.Arrays;
  9. import org.apache.hadoop.conf.Configuration;
  10. import org.apache.hadoop.fs.FSDataInputStream;
  11. import org.apache.hadoop.fs.FSDataOutputStream;
  12. import org.apache.hadoop.fs.FileStatus;
  13. import org.apache.hadoop.fs.FileSystem;
  14. import org.apache.hadoop.fs.LocatedFileStatus;
  15. import org.apache.hadoop.fs.Path;
  16. import org.apache.hadoop.fs.RemoteIterator;
  17. import org.junit.Before;
  18. import org.junit.Test;
  20. public class HdfsClientDemo {
  21.     
  22.     
  23.     public static void main(String[] args) throws Exception {
  24.         /**
  25.          * Configuration参数对象的机制：
  26.          *    构造时，会加载jar包中的默认配置 xx-default.xml
  27.          *    再加载 用户配置xx-site.xml  ，覆盖掉默认参数
  28.          *    构造完成之后，还可以conf.set("p","v")，会再次覆盖用户配置文件中的参数值
  29.          */
  30.         // new Configuration()会从项目的classpath中加载core-default.xml hdfs-default.xml core-site.xml hdfs-site.xml等文件
  31.         Configuration conf = new Configuration();
  32.         
  33.         // 指定本客户端上传文件到hdfs时需要保存的副本数为：2
  34.         conf.set("dfs.replication", "2");
  35.         // 指定本客户端上传文件到hdfs时切块的规格大小：64M
  36.         conf.set("dfs.blocksize", "64m");
  37.         
  38.         // 构造一个访问指定HDFS系统的客户端对象: 参数1:——HDFS系统的URI，参数2：——客户端要特别指定的参数，参数3：客户端的身份（用户名）
  39.         FileSystem fs = FileSystem.get(new URI("hdfs://172.31.2.38:9000/"), conf, "root");
  40.         
  41.         // 上传一个文件到HDFS中
  42.         fs.copyFromLocalFile(new Path("D:/install-pkgs/hbase-1.2.1-bin.tar.gz"), new Path("/aaa/"));
  43.         
  44.         fs.close();
  45.     }
  46.     
  47.     FileSystem fs = null;
  48.     
  49.     @Before
  50.     public void init() throws Exception{
  51.         Configuration conf = new Configuration();
  52.         conf.set("dfs.replication", "2");
  53.         conf.set("dfs.blocksize", "64m");
  54.         
  55.         fs = FileSystem.get(new URI("hdfs://172.31.2.38:9000/"), conf, "root");
  56.         
  57.     }
  58.     
  59.     
  60.     /**
  61.      * 从HDFS中下载文件到客户端本地磁盘
  62.      * @throws IOException 
  63.      * @throws IllegalArgumentException 
  64.      */
  65.     @Test
  66.     public void testGet() throws IllegalArgumentException, IOException{
  67.         
  68.         fs.copyToLocalFile(new Path("/test"), new Path("d:/"));
  69.         fs.close();
  70.         
  71.     }
  72.     
  73.     
  74.     /**
  75.      * 在hdfs内部移动文件\修改名称
  76.      */
  77.     @Test
  78.     public void testRename() throws Exception{
  79.         
  80.         fs.rename(new Path("/install.log"), new Path("/aaa/in.log"));
  81.         
  82.         fs.close();
  83.         
  84.     }
  85.     
  86.     /**
  87.      * 在hdfs中创建文件夹
  88.      */
  89.     @Test
  90.     public void testMkdir() throws Exception{
  91.         
  92.         fs.mkdirs(new Path("/xx/yy/zz"));
  93.         
  94.         fs.close();
  95.     }
  96.     
  97.     
  98.     /**
  99.      * 在hdfs中删除文件或文件夹
  100.      */
  101.     @Test
  102.     public void testRm() throws Exception{
  103.         
  104.         fs.delete(new Path("/aaa"), true);
  105.         
  106.         fs.close();
  107.     }
  108.     
  109.     
  110.     
  111.     /**
  112.      * 查询hdfs指定目录下的文件信息
  113.      */
  114.     @Test
  115.     public void testLs() throws Exception{
  116.         // 只查询文件的信息,不返回文件夹的信息
  117.         RemoteIterator<LocatedFileStatus> iter = fs.listFiles(new Path("/"), true);
  118.         
  119.         while(iter.hasNext()){
  120.             LocatedFileStatus status = iter.next();
  121.             System.out.println("文件全路径："+status.getPath());
  122.             System.out.println("块大小："+status.getBlockSize());
  123.             System.out.println("文件长度："+status.getLen());
  124.             System.out.println("副本数量："+status.getReplication());
  125.             System.out.println("块信息："+Arrays.toString(status.getBlockLocations()));
  126.             
  127.             System.out.println("--------------------------------");
  128.         }
  129.         fs.close();
  130.     }
  131.    
  133.     /**
  134.      * 读取hdfs中的文件的内容
  135.      * 
  136.      * @throws IOException
  137.      * @throws IllegalArgumentException
  138.      */
  139.     @Test
  140.     public void testReadData() throws IllegalArgumentException, IOException {
  142.         FSDataInputStream in = fs.open(new Path("/test.txt"));
  144.         BufferedReader br = new BufferedReader(new InputStreamReader(in, "utf-8"));
  146.         String line = null;
  147.         while ((line = br.readLine()) != null) {
  148.             System.out.println(line);
  149.         }
  151.         br.close();
  152.         in.close();
  153.         fs.close();
  155.     }
  157.     /**
  158.      * 读取hdfs中文件的指定偏移量范围的内容
  159.      * 
  160.      * 
  161.      * 
  162.      * @throws IOException
  163.      * @throws IllegalArgumentException
  164.      */
  165.     @Test
  166.     public void testRandomReadData() throws IllegalArgumentException, IOException {
  168.         FSDataInputStream in = fs.open(new Path("/xx.dat"));
  170.         // 将读取的起始位置进行指定
  171.         in.seek(12);
  173.         // 读16个字节
  174.         byte[] buf = new byte[16];
  175.         in.read(buf);
  177.         System.out.println(new String(buf));
  179.         in.close();
  180.         fs.close();
  182.     }
  184.     /**
  185.      * 往hdfs中的文件写内容
  186.      * 
  187.      * @throws IOException
  188.      * @throws IllegalArgumentException
  189.      */
  191.     @Test
  192.     public void testWriteData() throws IllegalArgumentException, IOException {
  194.         FSDataOutputStream out = fs.create(new Path("/zz.jpg"), false);
  196.         // D:\images\006l0mbogy1fhehjb6ikoj30ku0ku76b.jpg
  198.         FileInputStream in = new FileInputStream("D:/images/006l0mbogy1fhehjb6ikoj30ku0ku76b.jpg");
  200.         byte[] buf = new byte[1024];
  201.         int read = 0;
  202.         while ((read = in.read(buf)) != -1) {
  203.             out.write(buf,0,read);
  204.         }
  205.         
  206.         in.close();
  207.         out.close();
  208.         fs.close();
  210.     }
  212. }

 

练习：从一个文件里面不断地采集日志上传到hdfs里面

1.流程介绍

---启动一个定时任务

　　　　--定时探测日志原目录

　　　　--获取文件上传到一个待上传的临时目录

　　　　--逐一上传到hdfs目标路径，同时移动到备份目录里

--启动一个定时任务：

　　　　--探测备份目录中的备份数据是否已经超出，如果超出就删除

 

 主类为：

  1. import java.util.Timer;
  3. public class DataCollectMain {
  4.     
  5.     public static void main(String[] args) {
  6.         
  7.         Timer timer = new Timer();
  8.         //第一个为task类，第二个开始时间 第三个没隔多久执行一次
  9.         timer.schedule(new CollectTask(), 0, 60*60*1000L);
  10.         
  11.         timer.schedule(new BackupCleanTask(), 0, 60*60*1000L);
  12.         
  13.     }
  14.     
  16. }

CollectTask类：

这个类要继承TimerTask，重写run方法，主要内容就是不断收集日志文件

  1. package cn.edu360.hdfs.datacollect;
  3. import java.io.File;
  4. import java.io.FilenameFilter;
  5. import java.net.URI;
  6. import java.text.SimpleDateFormat;
  7. import java.util.Arrays;
  8. import java.util.Date;
  9. import java.util.Properties;
  10. import java.util.TimerTask;
  11. import java.util.UUID;
  13. import org.apache.commons.io.FileUtils;
  14. import org.apache.hadoop.conf.Configuration;
  15. import org.apache.hadoop.fs.FileSystem;
  16. import org.apache.hadoop.fs.Path;
  17. import org.apache.log4j.Logger;
  19. public class CollectTask extends TimerTask {
  21.     @Override
  22.     public void run() {
  23.         try {
  24.             // 获取配置参数
  25.             Properties props = PropertyHolderLazy.getProps();
  27.             // 获取本次采集时的日期
  28.             SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd-HH");
  29.             String day = sdf.format(new Date());
  30.             
  31.             File srcDir = new File("d:/logs/accesslog");
  32.             // 列出日志源目录中需要采集的文件
  33.             //里面传了一个文件过滤器，重写accept方法,return true就要
  34.             File[] listFiles = srcDir.listFiles(new FilenameFilter() {
  35.                 @Override
  36.                 public boolean accept(File dir, String name) {
  37.                     if (name.startsWith("access.log")) {
  38.                         return true;
  39.                     }
  40.                     return false;
  41.                 }
  42.             });
  43.             // 将要采集的文件移动到待上传临时目录
  44.             File toUploadDir = new File("d:/logs/toupload");
  45.             for (File file : listFiles) {
  46.                 
  47.                 //这里如果是 file.renameTo(toUploadDir)是不对的，因为会生成一个toupload的文件而不是文件夹
  48.                 //要用renameTo的话你要自己加上文件的新名字比较麻烦
  49.                 //用FileUtiles是对file操作的一些工具类
  50.                 //第一个目标文件，第二个路径，第三个是否存在覆盖
  51.                 FileUtils.moveFileToDirectory(file, toUploadDir, true);
  52.             }
  54.             // 构造一个HDFS的客户端对象
  55.             FileSystem fs = FileSystem.get(new URI("hdfs://hdp-01:9000"), new Configuration(), "root");
  56.             
  57.             File[] toUploadFiles = toUploadDir.listFiles();
  59.             // 检查HDFS中的日期目录是否存在，如果不存在，则创建
  60.             Path hdfsDestPath = new Path("/logs" + day);
  61.             if (!fs.exists(hdfsDestPath)) {
  62.                 fs.mkdirs(hdfsDestPath);
  63.             }
  65.             // 检查本地的备份目录是否存在，如果不存在，则创建
  66.             File backupDir = new File("d:/logs/backup" + day + "/");
  67.             if (!backupDir.exists()) {
  68.                 backupDir.mkdirs();
  69.             }
  71.             for (File file : toUploadFiles) {
  72.                 // 传输文件到HDFS并改名access_log_
  73.                 fs.copyFromLocalFile(new Path(file.getAbsolutePath()), new Path("/logs"+day+"/access_log_"+UUID.randomUUID()+".log"));
  75.                 // 将传输完成的文件移动到备份目录
  76.                 //注意这里依然不要用renameTo
  77.                 FileUtils.moveFileToDirectory(file, backupDir, true);
  78.             }
  80.         } catch (Exception e) {
  81.             e.printStackTrace();
  82.         }
  84.     }
  1. /**
  2.      * 读取hdfs中的文件的内容
  3.      * 
  4.      * @throws IOException
  5.      * @throws IllegalArgumentException
  6.      */
  7.     @Test
  8.     public void testReadData() throws IllegalArgumentException, IOException {
  10.         FSDataInputStream in = fs.open(new Path("/test.txt"));
  12.         BufferedReader br = new BufferedReader(new InputStreamReader(in, "utf-8"));
  14.         String line = null;
  15.         while ((line = br.readLine()) != null) {
  16.             System.out.println(line);
  17.         }
  19.         br.close();
  20.         in.close();
  21.         fs.close();
  23.     }
  25.     /**
  26.      * 读取hdfs中文件的指定偏移量范围的内容
  27.      * 
  28.      * 
  29.      * 作业题：用本例中的知识，实现读取一个文本文件中的指定BLOCK块中的所有数据
  30.      * 
  31.      * @throws IOException
  32.      * @throws IllegalArgumentException
  33.      */
  34.     @Test
  35.     public void testRandomReadData() throws IllegalArgumentException, IOException {
  37.         FSDataInputStream in = fs.open(new Path("/xx.dat"));
  39.         // 将读取的起始位置进行指定
  40.         in.seek(12);
  42.         // 读16个字节
  43.         byte[] buf = new byte[16];
  44.         in.read(buf);
  46.         System.out.println(new String(buf));
  48.         in.close();
  49.         fs.close();
  51.     }
  53.     /**
  54.      * 往hdfs中的文件写内容
  55.      * 
  56.      * @throws IOException
  57.      * @throws IllegalArgumentException
  58.      */
  60.     @Test
  61.     public void testWriteData() throws IllegalArgumentException, IOException {
  63.         FSDataOutputStream out = fs.create(new Path("/zz.jpg"), false);
  65.         // D:\images\006l0mbogy1fhehjb6ikoj30ku0ku76b.jpg
  67.         FileInputStream in = new FileInputStream("D:/images/006l0mbogy1fhehjb6ikoj30ku0ku76b.jpg");
  69.         byte[] buf = new byte[1024];
  70.         int read = 0;
  71.         while ((read = in.read(buf)) != -1) {
  72.             out.write(buf,0,read);
  73.         }
  74.         
  75.         in.close();
  76.         out.close();
  77.         fs.close();
  79.     }
  1.  

 

  2. }

BackupCleanTask类

  1. package cn.edu360.hdfs.datacollect;
  3. import java.io.File;
  4. import java.text.SimpleDateFormat;
  5. import java.util.Date;
  6. import java.util.TimerTask;
  8. import org.apache.commons.io.FileUtils;
  10. public class BackupCleanTask extends TimerTask {
  12.     @Override
  13.     public void run() {
  15.         SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd-HH");
  16.         long now = new Date().getTime();
  17.         try {
  18.             // 探测本地备份目录
  19.             File backupBaseDir = new File("d:/logs/backup/");
  20.             File[] dayBackDir = backupBaseDir.listFiles();
  22.             // 判断备份日期子目录是否已超24小时
  23.             for (File dir : dayBackDir) {
  24.                 long time = sdf.parse(dir.getName()).getTime();
  25.                 if(now-time>24*60*60*1000L){
  26.                     FileUtils.deleteDirectory(dir);
  27.                 }
  28.             }
  29.         } catch (Exception e) {
  30.             e.printStackTrace();
  31.         }
  33.     }
  35. }

 

hdfs中namenode中储存元数据（对数据的描述信息）是在内存中以树的形式储存的，并且每隔一段时间都会把这些元数据序列化到磁盘中。序列化的东西在磁盘中叫 fsimage文件。

元数据可能会很大很大，所以只能是定期的序列化

问题1：序列化的时候，发生了元数据的修改怎么办

答：namenode会把每次用户的操作都记录下来，记录成日志文件，存在edits日志文件中

其中edits日志文件也会像log4j滚动日志文件一样，当文件太大的时候会另起一个文件并改名字

问题2：当edits文件太多的时候，一次宕机也会花大量的时间从edits里恢复，怎么办

答：会定期吧edits文件重放fsimage文件，并记录edits的编号，把那些重放过的日志文件给删除。这样也相当于重新序列化了，

所以namenode并不会做这样的事情，是由secondary node做的，他会定期吧namenode的fsimage文件和edits文件下载下来

并把fsimage文件反序列化，并且读日志文件更新元数据，然后序列化到磁盘，然后把他上传给namenode。

这个机制叫做checkpoint机制

这里secondarynode 相当一一个小秘书

 

 

 

 

 

 额外知识点

注意，在windows里面不要写有些路径不要写绝对路径，因为程序放到linux下面可能会找不到，因此报错

 一般使用class加载器，这样当这个class加载的时候就会知道这个class在哪

类加载器的一些使用例子

比如我加载一个配置文件，为了避免出现绝对路径，我们可以是用类加载器

  1. 　　　　 Properties props = new Properties();
  2.         //加载配置文件，这样写的目的是为了避免在windows里出现绝对路径，用类加载器，再把文件传化成流
  3.         props.load(HdfsWordcount.class.getClassLoader().getResourceAsStream("job.properties"));

 

而对于一些功能性的类，我们最好在写逻辑的时候也不要直接去导入这个包，而是使用Class.forName

  1. //这样不直接导入这个包，直接用类加载器，是面向接口编程的一种思想。这里我并不是在开始import xxxx.Mapper,这里Mapper是一个接口，这里我用了多态
  2.         Class<?> mapper_class = Class.forName(props.getProperty("MAPPER_CLASS"));
  3.         Mapper mapper = (Mapper) mapper_class.newInstance();

 

单例模式

https://www.cnblogs.com/crazy-wang-android/p/9054771.html

只有个一实例，必须自己创建自己这个实例，必须为别人提供这个实例

 

 

 饿汉式单例：就算没有人调用这个class，他也会加载进去；

如对于一个配置文件的加载

  1. import java.util.Properties;
  3. /**
  4.  * 单例设计模式，方式一： 饿汉式单例
  5.  *
  6.  */
  7. public class PropertyHolderHungery {
  9.     private static Properties prop = new Properties();
  11.     static {
  12.         try {
  13.             //将一个文件prop.load(stram)  
  14.             //这里面如果传一个IO流不好，因为要用到绝对路径，使用了类加载器  这种不管有没有使用这个类都会加载
  15.             prop.load(PropertyHolderHungery.class.getClassLoader().getResourceAsStream("collect.properties"));
  16.         } catch (Exception e) {
  18.         }
  19.     }
  20.     public static Properties getProps() throws Exception {
  21.         return prop;
  22.     }
  24. }

 懒汉式：只有调用的时候才会有，但会有线程安全问题

  1. /**
  2.  * 单例模式：懒汉式——考虑了线程安全
  3.  * */
  4.  
  5. public class PropertyHolderLazy {
  7.     private static Properties prop = null;
  9.     public static Properties getProps() throws Exception {
  10.         if (prop == null) {
  11.             synchronized (PropertyHolderLazy.class) {
  12.                 if (prop == null) {
  13.                     prop = new Properties();
  14.                     prop.load(PropertyHolderLazy.class.getClassLoader().getResourceAsStream("collect.properties"));
  15.                 }
  16.             }
  17.         }
  18.         return prop;
  19.     }
  21. }

 

原文链接：http://www.cnblogs.com/wpbing/p/11233399.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Hadoop热门文章

  • Hive中的去重 ：distinct,group by与ROW_Number()窗口函数
  • HBASE的安装过程及运行HBASE程序的需要配置的内容
  • hadoop--Unable to load native-hadoop library for your platform解决方法
  • Hive sampling 语法之TABLESAMPLE用法理解
  • idea项目运行时报错提示Could not locate executable null\bin\winutils.exe in the Hadoop binaries
  • hadoop 无法访问50070端口问题解决
  • Hadoop运行时遇到java.io.FileNotFoundException错误的解决方法
  • KUDU数据导入尝试一：TextFile数据导入Hive,Hive数据导入KUDU
  • Hive安装
  • Hadoop 学习笔记 生态

    • Hadoop推荐文章

  • 基于Hive的大数据分析系统
  • HBase 在统一内容平台业务的优化实践
  • hive窗口分析函数使用详解系列一
  • 两台实体机器4个虚拟机节点的Hadoop集群搭建（Ubuntu版）
  • 在阿里云和腾讯云的轻量应用服务器上搭建Hadoop集群
  • Haproxy搭建 Web 群集实现负载均衡
  • HBase Compaction 原理与线上调优实践
  • Hive执行计划之hive依赖及权限查询和常见使用场景
  • 工良出品：包教会，Hadoop、Hive 搭建部署简易教程
  • Centos7系统编译Hadoop3.3.4

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号