草庐IT

Hadoop Map-Reduce的压缩最终输出文件

charles_wang888 2023-03-28 原文
 Hadoop 可以在作业的Configuration对象中通过设定一系列参数来改变作业的行为，比如，我们需要进行一个map-reduce作业，并且吧最终作业reduce过程的结果输出为压缩的格式，我们可以在一般的map-reduce上进行一些定制。

 

实现

还是以以前做的删选最高气温的例子为参照：

以前的例子可以见这个博文：http://supercharles888.blog.51cto.com/609344/878422

我们现在要求让结果输出为压缩格式，所以保持Map类(MaxTemperatureMapper）和Reduce类（MaxTemperatureReducer）不变，只要在Job类的Configuration作一些压缩的配置即可，见第45-49行所示：

  1. package com.charles.parseweather.compression; 
  2.  
  3.  
  4. import org.apache.hadoop.conf.Configuration; 
  5. import org.apache.hadoop.fs.Path; 
  6. import org.apache.hadoop.io.IntWritable; 
  7. import org.apache.hadoop.io.Text; 
  8. import org.apache.hadoop.io.compress.CompressionCodec; 
  9. import org.apache.hadoop.io.compress.GzipCodec; 
  10. import org.apache.hadoop.mapreduce.Job; 
  11. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
  12. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; 
  13. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
  14. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; 
  15.  
  16.  
  17. /** 
  18.  *  
  19.  * 
  20.  * Description: 这个类定义并且运行作业,压缩版 
  21.  * 
  22.  * @author charles.wang 
  23.  * @created May 24, 2012 5:29:12 PM 
  24.  * 
  25.  */ 
  26.  
  27. public class MaxTemperatureWithCompression { 
  28.  
  29.     /** 
  30.      * @param args 
  31.      */ 
  32.     public static void main(String[] args) throws Exception{ 
  33.         // TODO Auto-generated method stub 
  34.  
  35.          
  36.         if (args.length !=2){ 
  37.             System.err.println("Usage: MaxTemperature <input path> <output path>"); 
  38.             System.exit(-1); 
  39.         } 
  40.          
  41.         //创建一个Map-Reduce的作业 
  42.         Configuration conf = new Configuration(); 
  43.         conf.set("hadoop.job.ugi", "hadoop-user,hadoop-user"); 
  44.          
  45.         //在这里我们配置一些和压缩有关的参数 
  46.          
  47.         //我们设定reduce输出结果使用gzip压缩的形式 
  48.         conf.setBoolean("mapred.output.compress", true); 
  49.         conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class); 
  50.          
  51.          
  52.          
  53.         Job job = new Job(conf,"Get Maximum Weather Information with Compression! ^_^"); 
  54.          
  55.        
  56.          
  57.          
  58.          
  59.         //设定作业的启动类/  
  60.         job.setJarByClass(MaxTemperatureWithCompression.class); 
  61.          
  62.         //解析输入和输出参数，分别作为作业的输入和输出，都是文件 
  63.         FileInputFormat.addInputPath(job, new Path(args[0])); 
  64.         FileOutputFormat.setOutputPath(job, new Path(args[1])); 
  65.         
  66.         //配置作业，设定Mapper类，Reducer类 
  67.         job.setMapperClass(MaxTemperatureMapper.class); 
  68.         job.setReducerClass(MaxTemperatureReducer.class); 
  69.         job.setOutputKeyClass(Text.class); 
  70.        job.setOutputValueClass(IntWritable.class); 
  71.         
  72.         
  73.       
  74.          
  75.         System.exit(job.waitForCompletion(true)?0:1); 
  76.         
  77.          
  78.          
  79.          
  80.          
  81.  
  82.     } 
  83.  
  84. } 
  

要运行这个例子，我们需要给出输入文件，因为Hadoop系统可以根据输入文件的扩展名自动识别基本文件，所以我们创建目录结构，并且上传一个gzip格式的文件作为map-reduce过程的输入：

 

然后我们运行的main中传入HDFS的输入文件和输出目录：

 

当执行完成之后，我们就可以在HDFS文件系统中看到最终的输出结果了，正如我们所预期的，这个结果是个gzip格式的文件：

 

通过日志观察压缩输出文件过程

我们可以观察日志来更细粒度的观察整个过程：

namenode:

  1. 2012-05-31 13:11:08,621 INFO org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit: ugi=hadoop-user,hadoop-user ip=/192.168.40.16   cmd=open    src=/user/hadoop-user/compress-input/1901.gz    dst=null    perm=null 
  2. 2012-05-31 13:11:08,754 INFO org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit: ugi=hadoop-user,hadoop-user ip=/192.168.40.16   cmd=open    src=/user/hadoop-user/compress-input/1901.gz    dst=null    perm=null 
  3. 2012-05-31 13:11:08,758 INFO org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit: ugi=hadoop-user,hadoop-user ip=/192.168.40.16   cmd=mkdirs  src=/user/hadoop-user/compress-output/_temporary    dst=null    perm=hadoop-user:supergroup:rwxr-xr-x 
  4. 2012-05-31 13:11:08,853 INFO org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit: ugi=hadoop-user,hadoop-user ip=/192.168.40.16   cmd=open    src=/user/hadoop-user/compress-input/1901.gz    dst=null    perm=null 
  5. 2012-05-31 13:11:09,203 INFO org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit: ugi=hadoop-user,hadoop-user ip=/192.168.40.16   cmd=create  src=/user/hadoop-user/compress-output/_temporary/_attempt_local_0001_r_000000_0/part-r-00000.gz dst=null    perm=hadoop-user:supergroup:rw-r--r-- 
  6. 2012-05-31 13:11:09,238 INFO org.apache.hadoop.hdfs.StateChange: BLOCK* NameSystem.allocateBlock: /user/hadoop-user/compress-output/_temporary/_attempt_local_0001_r_000000_0/part-r-00000.gz. blk_-3869950436265612646_1016 
  7. 2012-05-31 13:11:09,292 INFO org.apache.hadoop.hdfs.StateChange: BLOCK* NameSystem.addStoredBlock: blockMap updated: 192.168.129.35:50010 is added to blk_-3869950436265612646_1016 size 29 
  8. 2012-05-31 13:11:09,686 INFO org.apache.hadoop.hdfs.StateChange: DIR* NameSystem.completeFile: file /user/hadoop-user/compress-output/_temporary/_attempt_local_0001_r_000000_0/part-r-00000.gz is closed by DFSClient_-356100022 
  9. 2012-05-31 13:11:09,692 INFO org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit: ugi=hadoop-user,hadoop-user ip=/192.168.40.16   cmd=listStatus  src=/user/hadoop-user/compress-output/_temporary/_attempt_local_0001_r_000000_0 dst=null    perm=null 
  10. 2012-05-31 13:11:09,695 INFO org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit: ugi=hadoop-user,hadoop-user ip=/192.168.40.16   cmd=mkdirs  src=/user/hadoop-user/compress-output   dst=null    perm=hadoop-user:supergroup:rwxr-xr-x 
  11. 2012-05-31 13:11:09,698 INFO org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit: ugi=hadoop-user,hadoop-user ip=/192.168.40.16   cmd=rename  src=/user/hadoop-user/compress-output/_temporary/_attempt_local_0001_r_000000_0/part-r-00000.gz dst=/user/hadoop-user/compress-output/part-r-00000.gz   perm=hadoop-user:supergroup:rw-r--r-- 
  12. 2012-05-31 13:11:09,699 INFO org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit: ugi=hadoop-user,hadoop-user ip=/192.168.40.16   cmd=delete  src=/user/hadoop-user/compress-output/_temporary/_attempt_local_0001_r_000000_0 dst=null    perm=null 
  13. 2012-05-31 13:11:09,703 INFO org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit: ugi=hadoop-user,hadoop-user ip=/192.168.40.16   cmd=delete  src=/user/hadoop-user/compress-output/_temporary    dst=null    perm=null 
  14. 2012-05-31 13:11:51,010 INFO org.apache.hadoop.hdfs.server.namenode.FSNamesystem.audit: ugi=hadoop-user,hadoop-user ip=/192.168.129.35  cmd=listStatus  src=/user/hadoop-user/compress-output   dst=null    perm=null 
 datanode:

  1. 2012-05-31 13:11:08,864 INFO org.apache.hadoop.hdfs.server.datanode.DataNode.clienttrace: src: /192.168.129.35:50010, dest: /192.168.40.16:6233, bytes: 74447, op: HDFS_READ, cliID: DFSClient_-356100022, srvID: DS-1002949858-192.168.129.35-50010-1337839176422, blockid: blk_-4455870079864415553_1015 
  2. 2012-05-31 13:11:09,248 INFO org.apache.hadoop.hdfs.server.datanode.DataNode: Receiving block blk_-3869950436265612646_1016 src: /192.168.40.16:6234 dest: /192.168.129.35:50010 
  3. 2012-05-31 13:11:09,283 INFO org.apache.hadoop.hdfs.server.datanode.DataNode.clienttrace: src: /192.168.40.16:6234, dest: /192.168.129.35:50010, bytes: 29, op: HDFS_WRITE, cliID: DFSClient_-356100022, srvID: DS-1002949858-192.168.129.35-50010-1337839176422, blockid: blk_-3869950436265612646_1016 
  4. 2012-05-31 13:11:09,283 INFO org.apache.hadoop.hdfs.server.datanode.DataNode: PacketResponder 0 for block blk_-3869950436265612646_1016 terminating 
  

我们在这里清楚的看到在目标目录下生成gzip格式的输出文件的整个过程,假定namenode第i行日志设为N(i)，datanode第i行日志设为D(i),则执行顺序为：

N1->N2->N3->N4->D1->N5->N6->D2->D3->D4->N7->N8...->N14,

其中N1->N4是namenode做一些准备工作，包括打开输入文件和创建输出目录及其临时子目录。

D1是datanode读取输入文件

N5,N6按照命名规则和配置中压缩文件的设定，创建输出文件到临时目录下（此时这个文件为空），然后用NameSystem吧这个块分配给datanode

D2-D4是datanode写最终reduce结果到被分配的块中。

N7-N14则是namenode吧输出文件的位置复制到命令行第二个参数指定的位置中，作为最终输出结果

 

 

 

 

Map-ReduceHadoopnbspspanclass压缩输出格式Hadoop大数据

有关Hadoop Map-Reduce的压缩最终输出文件的更多相关文章

  1. ruby - 使用 RubyZip 生成 ZIP 文件时设置压缩级别 - 2

    我有一个Ruby程序，它使用rubyzip压缩XML文件的目录树。gem。我的问题是文件开始变得很重，我想提高压缩级别，因为压缩时间不是问题。我在rubyzipdocumentation中找不到一种为创建的ZIP文件指定压缩级别的方法。有人知道如何更改此设置吗？是否有另一个允许指定压缩级别的Ruby库？ 最佳答案 这是我通过查看ruby​​zip内部创建的代码。level=Zlib::BEST_COMPRESSIONZip::ZipOutputStream.open(zip_file)do|zip|Dir.glob("**/*")d

  2. ruby - 其他文件中的 Rake 任务 - 2

    我试图在一个项目中使用rake，如果我把所有东西都放到Rakefile中，它会很大并且很难读取/找到东西，所以我试着将每个命名空间放在lib/rake中它自己的文件中，我添加了这个到我的rake文件的顶部:Dir['#{File.dirname(__FILE__)}/lib/rake/*.rake'].map{|f|requiref}它加载文件没问题，但没有任务。我现在只有一个.rake文件作为测试，名为“servers.rake”，它看起来像这样:namespace:serverdotask:testdoputs"test"endend所以当我运行rakeserver:testid时

  3. ruby-on-rails - 在 Rails 中将文件大小字符串转换为等效千字节 - 2

    我的目标是转换表单输入，例如“100兆字节”或“1GB”，并将其转换为我可以存储在数据库中的文件大小(以千字节为单位)。目前，我有这个:defquota_convert@regex=/([0-9]+)(.*)s/@sizes=%w{kilobytemegabytegigabyte}m=self.quota.match(@regex)if@sizes.include?m[2]eval("self.quota=#{m[1]}.#{m[2]}")endend这有效，但前提是输入是倍数(“gigabytes”，而不是“gigabyte”)并且由于使用了eval看起来疯狂不安全。所以，功能正常，

  4. ruby-on-rails - Rails 3 中的多个路由文件 - 2

    Rails2.3可以选择随时使用RouteSet#add_configuration_file添加更多路由。是否可以在Rails3项目中做同样的事情？ 最佳答案 在config/application.rb中:config.paths.config.routes在Rails3.2(也可能是Rails3.1)中，使用:config.paths["config/routes"] 关于ruby-on-rails-Rails3中的多个路由文件，我们在StackOverflow上找到一个类似的问题

  5. ruby - 将差异补丁应用于字符串/文件 - 2

    对于具有离线功能的智能手机应用程序，我正在为Xml文件创建单向文本同步。我希望我的服务器将增量/差异(例如GNU差异补丁)发送到目标设备。这是计划:Time=0Server:hasversion_1ofXmlfile(~800kiB)Client:hasversion_1ofXmlfile(~800kiB)Time=1Server:hasversion_1andversion_2ofXmlfile(each~800kiB)computesdeltaoftheseversions(=patch)(~10kiB)sendspatchtoClient(~10kiBtransferred)Cl

  6. ruby - 如何将脚本文件的末尾读取为数据文件(Perl 或任何其他语言) - 2

    我正在寻找执行以下操作的正确语法(在Perl、Shell或Ruby中):#variabletoaccessthedatalinesappendedasafileEND_OF_SCRIPT_MARKERrawdatastartshereanditcontinues. 最佳答案 Perl用__DATA__做这个:#!/usr/bin/perlusestrict;usewarnings;while(){print;}__DATA__Texttoprintgoeshere 关于ruby-如何将脚

  7. ruby - 使用 Vim Rails，您可以创建一个新的迁移文件并一次性打开它吗？ - 2

    使用带有Rails插件的vim，您可以创建一个迁移文件，然后一次性打开该文件吗？textmate也可以这样吗？ 最佳答案 你可以使用rails.vim然后做类似的事情::Rgeneratemigratonadd_foo_to_bar插件将打开迁移生成的文件，这正是您想要的。我不能代表textmate。 关于ruby-使用VimRails，您可以创建一个新的迁移文件并一次性打开它吗？，我们在StackOverflow上找到一个类似的问题： https://sta

  8. ruby - 检查 "command"的输出应该包含 NilClass 的意外崩溃 - 2

    为了将Cucumber用于命令行脚本，我按照提供的说明安装了arubagem。它在我的Gemfile中，我可以验证是否安装了正确的版本并且我已经包含了require'aruba/cucumber'在'features/env.rb'中为了确保它能正常工作，我写了以下场景:@announceScenario:Testingcucumber/arubaGivenablankslateThentheoutputfrom"ls-la"shouldcontain"drw"假设事情应该失败。它确实失败了，但失败的原因是错误的:@announceScenario:Testingcucumber/ar

  9. ruby - 通过 erb 模板输出 ruby​​ 数组 - 2

    我正在使用puppet为ruby​​程序提供一组常量。我需要提供一组主机名，我的程序将对其进行迭代。在我之前使用的bash脚本中，我只是将它作为一个puppet变量hosts=>"host1,host2"我将其提供给bash脚本作为HOSTS=显然这对ruby​​不太适用——我需要它的格式hosts=["host1","host2"]自从phosts和putsmy_array.inspect提供输出["host1","host2"]我希望使用其中之一。不幸的是，我终其一生都无法弄清楚如何让它发挥作用。我尝试了以下各项:我发现某处他们指出我需要在函数调用前放置“function_”……这

  10. Ruby 写入和读取对象到文件 - 2

    好的，所以我的目标是轻松地将一些数据保存到磁盘以备后用。您如何简单地写入然后读取一个对象？所以如果我有一个简单的类classCattr_accessor:a,:bdefinitialize(a,b)@a,@b=a,bendend所以如果我从中非常快地制作一个objobj=C.new("foo","bar")#justgaveitsomerandomvalues然后我可以把它变成一个kindaidstring=obj.to_s#whichreturns""我终于可以将此字符串打印到文件或其他内容中。我的问题是，我该如何再次将这个id变回一个对象？我知道我可以自己挑选信息并制作一个接受该信

随机推荐