Hadoop in action 第45678章

第四五章
   MapReduce基础
       实例
           使用专利局的数据
           开发最好基于一个模板
           单个类完整定义每个MapReduce作业，Mapper和Reducer是自身静态类
           在执行期间，采用不同的jvm的各类节点复制并运行Mapper和Reducer而其他的作业仅在客户机上执行
       mr框架模板代码
           框架的核心在run函数中，也称为driver。
               实例化，配置并传递一个JobConf对象命名的作业给JobClient.runJob()来启动
           JobClient类与JobTracker通信让作业在集群中启动。JobConf对象保持了作业运行的全部参数
           -D mapred.reduce.tasks=0 放弃了reduce段计算
           -conf $configFile
           -D JobConf属性赋值
           -fs <local|namenode:port>指定一个namenode
           -jt <local|jobtracker:port>指定一个JobTracker
           MapReduceBase是一个小类。
               configure() close() 提供非操作性实现
               建立map|reduce任务,清除map|reduce任务

编写第一步是了解数据流。map的输入，map的输出，reduce的输入，reduce的输出
           选择InputFormat，设定"key.value.separator.in.input.line"属性值
       统计
           1分配型{最大值函数，最小值函数，总计函数，count函数}
           2代数型{平均值函数，方差函数}
           3全集型{中值函数，K个最大/最小的函数}
       脚本语言
           流式处理数据
       combiner带来的性能提升
           map阶段的输出，键的hash并非均匀分布，会造成某台机器的过重
           combiner的目的在于减少map输出在网络上的io负载，和reduce节点的压力。使得在每个map上，每个键仅有一个结果会参与洗牌
           combiner在数据的转换上和reduce必须等价，有无都不会改变reduce的最终输出
           Combiner是否带来性能提升不是必然的，应该通过测试获得确切答案
           Top K问题，Combiner提取每个map输出的Top K，Reduce合并多个Top K
           流量问题，Combiner累加每个map输出的每小时流量，Reduce再次合并多个文件中的某小时流量
           稀疏向量内乘
           时序处理，线性滤波器中务必使用combiner
           乘积
           虚构方言

MapReduce高级
       顺序链接
           当JobClient.runJob()运行到作业结尾处被组织时，MapReduce作业的链接会在一个作业之后调用另一个作业的driver
           每个作业的driver必须创建一个新的jobConf对象，并将其输入路径设置为前一个作业的输出路径。
           可以在最后阶段删除在链上每个阶段生成的中间数据
       多源合并链接
           A-》Result
           B-》Result
           Result[]->C

通过Job和JobConf来管理非线性的依赖
               Job对象的实例化可以通过传递一个JobConf对象到作业的构造函数来实现
               Jobx.addDependingJob(y)作业y完成之后作业x才启动
               JobControl会负责管理并监视作业的执行。
               JobControl.addJob(x)添加作业
               JobControl.run()生成一个线程来提交作业并监视其执行
       预处理和后处理
           预处理，清洗数据
           ChainMapper.addMapper() ChainReducer.addMapper()来组合预处理和后处理
           Map+|Reduce|Map+
           全部预处理和后处理不会生成中间文件，减少IO

driver：
Configuration conf = getConf();
JobConf job= new JobConf(conf);

job.setJobName("ChainJob");

job.setInputFormat(TextInputFormat.class);
               job.setOutputFormat(TextOutputFormat.class);
               FileInputFormat.setInputPaths(job,in);
               FileOutputFormat.setOutputPath(job,out);

JobConf map1Conf = new JobConf(false);
               JobConf map2Conf = new JobConf(false);
               JobConf reduceConf = new JobConf(false);
               JobConf map3Conf = new JobConf(false);
               JobConf map4Conf = new JobConf(false);

               ChainMapper.addMapper(job,Map1.class
                               ,LongWritable.class,Text.class,Text.class,Text.class
                               ,true,map1Conf);
               ChainMapper.addMapper(job,Map2.class
                               ,Text.class,Text.class,LongWritable.class,Text.class
                               ,true,map2Conf);
               ChainReducer.setReducer(job,Reducer.class
                               ,LongWritable.class,Text.class,Text.class,Text.class
                               ,true,reduceConf);
               ChainReducer.addMapper(job,Map3.class
                               ,Text.class,Text.class,LongWritable.class,Text.class
                               ,true,map3Conf);
               ChainReducer.addMapper(job,Map4.class
                               ,LongWritable.class,Text.class,LongWritable.class,Text.class
                               ,true,map4Conf);

JobClient.runJob(job);

add*函数有一个boolean参数 byValue。
标准的mapper计算结果，键值对会写入磁盘，等待洗牌进入另外一个节点。
true 值传递，发送键值对副本

多个Mapper相链，在一个进程内完全可以引用传递。
Map1{OutputCollector.collect(k,v);}直接传递给Map2{map(kv);}
但是API有一个约定，collect(k,v)不会改变kv的值，也就是说，map2的map方法不能改变输入数据的值

连接不同来源的数据
           两个相关联数据集的内联外联
           reduce阶段联接，datajoin是联接的通用框架，
               datasource
               tag
                   无状态的读入单个记录，会丢失记录的类型。标记这个记录会确保特定的元数据一直跟随记录。
                   对于联接来说，会为记录标记数据源。
               groupkey
                   定位是联接键

重分区排序-合并联结
在map阶段将一个记录标记 groupkey 和 tag {groupkey，TaggedMapOutput(tag,Text)}
在分区和洗牌阶段相同的groupkey 组成一组在相同的groupkey上调用reduce

在reduce阶段相同联接键的记录会被一起处理
reduce会得到n部分数据输入，N部分相同groupkey不同数据源
reduce将N部分输入进行完全交叉乘积(同数据源不相乘)，并输出乘积结果{不同的数据源中有相同groupkey的一个记录}

乘积结果进入reduce阶段的combin函数。combine函数决定了是内联还是外联，还是其他
内联：丢弃所有不含全部tag的的合并结果。比如没有订单的客户，{没有订单数据源}

DataJoinMapperBase
                   封装记录

                   Text generateInputTag(String inputFile);
                   在map任务开始时，为这次map任务的所有记录生成一个tag。其返回值保存在this.inputTag

然后调用DataJoinMapperBase的map函数

TaggedMapOutput generateTaggedMapOutput(object value);
                   map函数调用这个。
                       TaggedWritable ret= new TaggedWritable(value);
                       ret.setTag(this.inputTag);
                       return ret;

Text generateGroupKey(TaggedMapOutput record);
map函数再调用这个.
从记录中提取groupkey

map函数最后output.colloect(groupkey,taggedOutput);

map之后进入分区洗牌
               DataJoinReducerBase
                   reduce 完全交叉乘积M个数据源其下的任意个记录
                   combine(object[] tags,object[] values) 过滤。
                       其中两个数组的长度必须相同，就是每个数据源各取一个
                       如果tags的长度少于分析的数据源，内联不成功

reduce阶段输出<groupkey,combine()>
TaggedMapOutput
getTag()，getData()

基于DistributedCache的复制联接
               reduce联接技术灵活，但效率低下。
                   map阶段仅仅组合数据，reduce阶段接受了大量的数据，但是大量的数据又被丢弃

               map阶段去除不必要的数据，最好在map阶段就完成联接

问题在于多个数据源中同一个groupkey的记录不能同时出现。
               实际情况是联接数据时，只有一个数据源是大集合，其他数据源往往小得多
                   当较小的数据源可以装入内存时，我们可以将其复制到所有的mapper，并在map阶段完成联接
               Hadoop自身有一个分布式缓存的机制，将文件(mapper所需的背景文件)分布到集群所有点上
               DistributedCache
                   1配置作业时DistributedCache.addCacheFile()设定要分布的背景文件
                       JobTracker在每个TaskTracker中创建本地副本
                   2TaskTracker在map阶段 DistributedCache.getLocalCacheFiles()获取文件路径

Configuration conf = getConf();
JobConf job = new JobConf(conf,$DataJoin.class);

DistributedCache.addCacheFile(new Path(args[0]).toUri,conf);

Path in = new Path(args[1]);
                   Path out = new Path(args[2]);
                   FileInputFormat.setInputPaths(job,in);
                   FileInputFormat.setOutputPaths(job,out);
                   job.setInputFormat(KeyValueInputFormat.class);
                   job.setOutputFormat(TextOutputFormat.class);
                   job.set("key.value.separator.in.input.line",",");

job.setName("Join in Mapper");
job.setMapperClass($MapperClass);
job.setNumReduceTasks(0);

JobClient.runJob(job);

mapper
public static MapperClass extends MapReduceBase implements Mapper<Text,Text,Text,Text>{

private HashMap<String,String> joinData = new HashMap<String,String>();

public void configure(JobConf conf){
Path[] cacheFiles = DistributedCache.getLocalCachesFiles(conf);
if (cacheFiles == null||cacheFiles.length ==0){return ;}

String line;
                           String[] tokens;
                           Reader _reader=new FileReader();
                           BufferedReader reader = new BufferedReader(_reader);

for(line = reader.readLine();line!=null;line=reader.readLine()){
                               tokens = line.split(",",2);
                               joinData.put(tokens[0],tokens[1]);
                           }
                           finally{
                               reader.close();
                               _reader.close();
                           }
                       }

public void map(Text key,Text value,OutputCollector<Text,Text> output
                                       ,Reporter reporter){
                           String joinValue = joinData.get(key.toString());
                           if (joinValue == null) {return ;}

output.colloect(key,new Text(value.toString()+","+joinValue));
}

public void close(){}
}

半联接，map过滤，reduce联接
               复制联接有一个限制是背景文件要足够小到装入内存
               过滤出一个小的子集然后将小的子集作为背景文件分布出去
               如果满足查询条件的子集仍然大于内存是布隆过滤算法。

布隆过滤算法

bloom filter是一个数据集的摘要
           bloom算法对于过滤不会漏报，可能会误报。
               当bloom.contains() => false不包含就是不包含，
               当bloom.contains() => true 包含时，有可能不包含
               一般应用中表现为数据量大小已知，误报率已知，设计位数组大小和三类函数个数

算法优势在于数据结构的大小固定，处理的数据量不会改变bloom数据结构的大小

第六章编程实践
       绝技调试
           要提防程序正确，数学公式不正确的错误
               检查目标值是否符合预期，检查过程值

           检查引用统计的完整的性
           引入回归测试，专注比较前后的差异
               保存map阶段的输出
           计数使用long
           使用计数器插桩。
               Reporter.incrCount()
           跳过坏记录
               Hadoop对硬件的恢复机制无法应对坏记录造成的软件失败
               当有一个skipping模式。这个模式启动后，TaskTracker跟踪失败区域，重启后，忽略坏记录区域
               SkipBadRecords.setMapperMaxSkipRecords(Configuration conf,long maxSkipRecs);
               SkipBadRecords.setReducerMaxSkipRecords(Configuration conf,long maxSkipRecs);
               通过二分法确定跳读的区域
               JobConf.setMaxMapAttempts()
               JobConf.setMaxReduceAttempts()这两个参数可以增加在二分测试中重试次数

被跳读的记录写入HDFS TaskTrackerHost:HOME/_log/skip 目录下
bin/hadoop fs -text hadoopzip

IsolationRunner重建任务
               配置keep.failed.task.files=true
               每个TaskTracker保存所有必要的数据来重新运行失效的任务
               当作业失效，使用TaskTracker的web页面定位失效的节点，作业id和任务的重试Id。
               登录到失败的TaskTracker,进入work目录
               $local_dir/taskTracker/jobcache/$job_id/$attempt_id/work
                   attempt_id以attempt_作前缀
                   $local_dir 是属性mapred.local.dir设置的
               在work目录中通过IsolationRunner使用以前相同的输入来重新运行
               bin/hadoop org.apache.hadoop.mapred.IsolationRunner ../job.xml

通过设置export HADOOP_OPTS="-agentlib:jdwp=transport=dt_socket,server=Y,address=8000"
TaskTracker 的jvm在8000端口上侦听调试器

使用 jdb -attach 8000

日志和监控
           日志在每个TaskTrackerHost:home目录下的logs目录中
               NameNode
               JobTracker
               TaskTracker

在logs目录下userlog会记录用户输出
           Reporter.setStatus可以实时传送状态
           JobTracker 的web控制台
               http://JobTrackerHost:50030/jobtracker.jsp
               跟踪作业Id。job_集群启动的时间戳_一个自增作业号
               JobConf。setName 作业名字

bin/hadoop job -kill job_id
       性能调优
           使用集群的线性扩展增加吞吐量
           提升单机性能，节省硬件

使用combiner减少网络流量
               减少了map和reduce两个阶段间洗牌的数量
           减少输入数据量
               减少样本大小，只降低精度，不影响准确性
               只使用有关字段
                   列数据库
           使用压缩
               mapred.compress.map.output=boolean
               mapred.map.output.compression.codec = class
               推荐使用hadoop专用的序列文件传递数据，序列文件将每个文档视为一条记录
                   SequenceFileOutputFormat

文件作为一个整体来解压缩就毁掉了并行性
序列文件兼顾了压缩和并行。添加同步标识来表示拆分的边界

压缩格式有RECORD和BLOCK。
记录压缩每条记录被分别压缩
块压缩记录中的块会被一起压缩得到更高的压缩率

FileOutputFormat.setCompressOutput()
FileOutputFormat.setOutputCompressorClass()让输出使用特定的算法实现

driver
               conf.setOutputFormat(SequenceFileOutputFormat);
               SequenceFileOutputFormat.setOutputCompressionType(conf,CompressionType.Block);
               FileOutputFormat.setCompressOutput(conf,true);
               FileOutputFormat.setOutputCompressorClass(conf,GzipCodec.class);

重用JVM
TaskTracker在独立的jvm中运行map或reduce任务。
改为一个jvm多次运行map或reduce任务。

JobConf.setNumTaskToExecutePerJvm(int);
mapred.jbo.reuse.jvm.num.tasks

猜测执行
hadoop注意到任务变慢，就会再次安排一个相同的任务
重构算法

第七八章手册与运维
       向任务传递参数
           在driver中配置JobConf
           可以用一个唯一的名字和值来表示一个自定义的参数
           JobConf会传递到所有的TaskTracker

任务初始化时会调用Mapper/Reducer.configure(JobConf){}得到自己定义的参数

       多个输出文件
           MultipleOutputFormat 将相似的记录结组为不同的数据集
               需要扩展，实现generateFileNameForKeyValue(),决定子文件名
               根据map的输出keyvalue决定这些keyvalue[] 如何划分成多个文件

           MultipleOutputs
               不会为每对keyvalue判断出文件名，而是创建多个OutputCollector，每个Collector都有自己键值对类型
               这样一分数据输入，两份输出
               输出文件以在driver中定义的名字为前缀
               但是到reduce阶段只有默认的part-00000才会被传递

driver
MultipleOutputs.addNamedOutput(JobConf,"name1",TextOutputFormat.class,NullWritable.class,Text.class);

mapper
                   MultipleOutputs mos;
                   void configure(JobConf job){
                       this.mos = new MultipleOutputs(job);
                   }
                   void map(Key,Value,OutputCollector<NullWritable,Text> output,reporter){
                       OutputCollector coll = this.mos.getCollector($value.parse,reporter);

                       coll.colloect(format1Key,format1Value);
                       coll.colloect(format2Key,format2Value);

}
void close(){this.mos.close();}

以数据库作为数据源
从数据库读数据性能不理想，因为数据向计算移动，有违hadoop的设计初衷

DBOutputFormat
       保持输出顺序
           reduce的输入按键来排序，
           简单的计算使得输出也依序输出
           如果不需求顺序可以reduce阶段为0，直接使用map的输出，分区和洗牌是为reduce存在。
           没有reduce，分区和洗牌也不存在

如果想要part-00000 < part-00001 < part-00002 这样的顺序全依赖map后reduce前的Partitioner
           Partitioner的任务是为每个key分配{hash(key)}一个reduce。这样相同的键的所有记录都在一个reduce中被计算
           默认的散列函数会使得reduce计算量相当，但洗牌过程的网络连接过于复杂 IO增大
           使用自定义的散列函数，是key相对集中，但是有可能负载不平衡因为key背后的记录就不平衡，需要commbine合并一下
       配置集群参数
           dfs.name.dir      NameNode节点，存储HDFS的元数据        /home/hadoop/dfs/name
           dfs.data.dir      DataNode节点,存储HDFS的块文件         /home/hadoop/dfs/data
                              如果有多个磁盘，应该再每个磁盘上
                              建立一个目录
           fs.trash.interval 文件被删除，缓冲多少分钟              0是不启用
           mapred.system.dir HDFS系统中存储共享Mapreduce文件的目录 /home/hadoop/mapred/system
           mapred.local.dir TaskNode节点存储临时数据的目录       。。。。。

           mapred.tasktracker.
           map|reduce.tasks.
           maximum           TaskTracker，这个节点上可以同时运行的
                               map和reduce最大值
           hadoop.tmp.dir    hadoop临时目录                        /home/hadoop/tmp

           dfs.datanode.du.
           reserved          DataNode节点,应该具备的最小空闲空间   1073741824

           mapred.child.java.
           opts              每个子任务的堆栈大小                  -Xmx512m

           mapred.reduce.
           tasks             一个作业的reduce任务数

检查HDFS
           bin/hadoop fsck / [-openwrite]
           不健康的文件。过度复制复制不足未复制损坏的失踪的
           但HDFS是自我修复的。过度复制复制不足未复制不足为虑
           使用-delete参数删除损坏的失踪的文件
           使用-openwrite参数会统计正在写入的文件
           使用-files -blocks -locations -racks。每个后续选项都需要前面选项存在
               -files    检查一个文件打印一行信息
               -blocks   检查一个块打印一行信息
               -location 块副本的位置
               -racks    机架名字

DataNode节点
           bin/hadoop dfsadmin -report
           NameNode节点
           bin/hadoop dfsadmin -metasave filename
               将一部分namenode的元数据保存到日志目录下的命名文件中
               包括等待复制的块，正在复制的块，等待删除的块
               统计摘要

bin/hadoop fs -chmod
           bin/hadoop fs -chown
           bin/hadoop fs -chgrp
       管理datanode
           移除
           单个移除或杀死
           批量退役。确保所有块在剩余的机器上仍有足够的副本
               在namenode的本地文件系统生成一个空的排除文件
               让dfs.hosts.exclude参数在启动时指向空的排除文件
               有移除需要时，逐行列出ip:port
               执行 bin/hadoop dfsadmin -refreshnodes
               NameNodeshang 日志出现Decommission complete for node 172.16.1.55:50010
               下架机器进行维护

节点上线
               安装hadoop
               配置参数
               启动守护进程 bin/hadoop datanode。这样此节点自动加入到集群中
               在namenode节点上 conf/slaves文件添加这个节点

新节点的加入，对于集群来说数据分配是不平衡的
bin/start-balance.sh进行数据平衡调节

平衡调节适合在作业稀疏时进行
           管理NameNode
               NameNode单独部署，大内存，raid
               减轻负担，增加数据块大小，但并行也被降低
               具体大小根据业务场景定

dfs.block.size

10台机器以上机器NameNode和snn应该分开
不是namenode失败时的备份，虽然可以扮演此角色
snn定时清理NameNode上的文件系统状态信息，使之紧凑，nameno更有效率

NameNode使用FsImg EditLog两个文件管理文件系统状态信息。
                       FsImg文件系统在一些检查点上的快照
                       EditLog是检查点之后的增量修改。
                   NameNode启动时，两个文件合并形成新快照。
                   两个文件都是不运行的描述，运行增量改变都在内存中，及时响应查询

当集群工作量大，editlog变大，重启变慢。
snn合并fsimg和editlog，形成新的快照，让NameNode专注作业。

snn作为一个检查点服务器更合适

bin/start-dfs.sh
bin/start-mapred.sh

多用户作业调度
hadoop默认是fifo调度。很容易一个大的作业阻塞众多小的作业
本办法是在一个hdfs集群上架构多个mapreduce集群，物理并行。数据可能不在mapreduce集群上的数据节点

公平调度器
                   池{一定数量的map slot，一定数量的reduce slot}
                   每个作业都标记了池的归属
                   当task slot可用时，调度器首先满足这些最低限度的保障
                   slot再在作业间共享，使得每个作业大致同等的计算资源

contrib/faircheduler/目录下，将jar放入hadoop的lib目录下
                   hadoop-site.xml
                       mapred.JobTracker.taskscheduler = org.apache.hadoop.mapred.FairScheduler
                       mapred.fairscheduler.allocation.file=池的定义文件 pool.xml{池的名字和容量}
                       mapred.fairscheduler.poolnameproperty=pool.name jobconf属性，调度器通过这个来决定使用哪个池
                           一般将其设置一个新属性：pool.name。这个属性的默认值为${user.name}
                           调度器自动赋予每个用户独立的池
                       pool.name=${user.name}

Hadoop in action 第45678章相关推荐

Hadoop In Action 第四章(1)
第四章编写基本的MapReduce程序本章涵盖了: 用Hadoop处理数据集,以专利数据为例一个MapReduce程序的基本结构基本的MapReduce程序,以数据统计为例 Hadoop的流A ...
Hadoop快速入门——第三章、MapReduce案例(字符统计)
Hadoop快速入门--第三章.MapReduce案例目录环境要求: 1.项目创建: 2.修改Maven 3.编码 4.本地文件测试 5.修改[Action]文件(修改测试文件路径) 6.导出ja ...
《Hadoop权威指南》第二章关于MapReduce
<Hadoop权威指南>第二章关于MapReduce 目录使用Hadoop来数据分析横向扩展注:<Hadoop权威指南>重点学习摘要笔记 1. 使用Hadoop来数据分 ...
Hadoop专业解决方案-第1章大数据和Hadoop生态圈
一.前言: 非常感谢Hadoop专业解决方案群:313702010,兄弟们的大力支持,在此说一声辛苦了,经过两周的努力,已经有啦初步的成果,目前第1章大数据和Hadoop生态圈小组已经翻译完成,在此 ...
Hadoop快速入门——第四章、zookeeper安装
Hadoop快速入门--第四章.zookeeper安装压缩包下载地址:[https://download.csdn.net/download/feng8403000/85227883] 目录 1.上 ...
《精通Hadoop》：第 1 章　Hadoop 2.X
第 1 章 Hadoop 2.X "没有什么是不能通过搜索引擎或互联网找到的." --埃里克·施密特,谷歌执行主席在大规模.高并行和分布式计算处理这个行业中,Hadoop实际上是 ...
《Gans in Action》第一章对抗神经网络介绍
此为<Gans in Action>(对抗神经网络实战)第一章读书笔记 Chapter 1. Introduction to GANs 对抗神经网络介绍 This chapter cove ...
Hadoop基础教程-第12章 Hive：进阶（12.3 HiveServer2）（草稿）
第12章 Hive:进阶 12.3 HiveServer2 12.3.1 HiveServer1 HiveServer是一种可选服务,允许远程客户端可以使用各种编程语言向Hive提交请求并检索结果.H ...
Hadoop基础教程-第10章 HBase：Hadoop数据库（10.1 NoSQL介绍）（草稿）
第10章 HBase:Hadoop数据库 10.1 NoSQL介绍 10.1.1 NoSQL简介随着互联网技术(互联网+,物联网)发展,特别是大数据时代到来,我们需要存储处理更多数据,这种需求远远超 ...

Hadoop in action 第45678章

Hadoop in action 第45678章相关推荐

最新文章

热门文章