标签:
Hadoop 由 Apache Software Foundation 公司于 2005 年秋天作为Lucene的子项目 Nutch的一部分正式引入。它受到最先由 Google Lab 开发的 Map/Reduce 和 Google File System(GFS) 的启发。Yahoo! 是最主要源代码贡献者, 贡献了大约80%的代码,Powerset写的HBase, Facebook 写的Hive都是Hadoop上很重要的子项目。Hadoop的使用异常广泛,凡是涉及大数据处理的互联网公司几乎都使用Hadoop,已知为接近150家的大型组织实际使用: Yahoo!, Amazon, EBay, AOL, Google, IBM, Facebook, Twitter, Baidu, Alibaba, Tencent。在这里可以看到一些使用Hadoop的公司。
Hadoop目标可以概括为:可扩展性: Petabytes (1015 Bytes) 级别的数据量, 数千个节点,经济性: 利用商品级(commodity)硬件完成海量数据存储和计算,可靠性: 在大规模集群上提供应用级别的可靠性。
Hadoop包含两个部分:
1、HDFS
HDFS即Hadoop Distributed File System(Hadoop分布式文件系统),HDFS具有高容错性,并且可以被部署在低价的硬件设备之上。HDFS很适合那些有大数据集的应用,并且提供了对数据读写的高吞吐率。HDFS是一个master/slave的结构,就通常的部署来说,在master上只运行一个Namenode,而在每一个slave上运行一个Datanode。
HDFS支持传统的层次文件组织结构,同现有的一些文件系统在操作上很类似,比如你可以创建和删除一个文件,把一个文件从一个目录移到另一个目录,重命名等等操作。Namenode管理着整个分布式文件系统,对文件系统的操作(如建立、删除文件和文件夹)都是通过Namenode来控制。
HDFS的结构图中可以看出,Namenode,Datanode,Client之间的通信都是建立在TCP/IP的基础之上的。当Client要执行一个写入的操作的时候,命令不是马上就发送到Namenode,Client首先在本机上临时文件夹中缓存这些数据,当临时文件夹中的数据块达到了设定的Block的值(默认是64M)时,Client便会通知Namenode,Namenode便响应Client的RPC请求,将文件名插入文件系统层次中并且在Datanode中找到一块存放该数据的block,同时将该Datanode及对应的数据块信息告诉Client,Client便这些本地临时文件夹中的数据块写入指定的数据节点。
HDFS采取了副本策略,其目的是为了提高系统的可靠性,可用性。HDFS的副本放置策略是三个副本,一个放在本节点上,一个放在同一机架中的另一个节点上,还有一个副本放在另一个不同的机架中的一个节点上。
2、MapReduce
有另一种分布式计算框架MPI,它在很多的问题上实现起来比Map/Reduce更方便,比如带迭代的机器学习的模型,但是个人还是要提倡Map/Reduce。原因是Map/Reduce模型中Map之间是相互独立的,因为相互独立,使得系统的可靠性大大提高了。比如一个任务需要1000个结点共同完成,在MPI中需要这1000个结点协同来完成这个任务,结点间可能有通信,数据交换。如果你有一个结点发生问题,整个任务就会失败,当然你也可能有一些容错的处理可以让任务继续算。但在海量数据情况下,比如你需要四位数的服务器运算一个任务,而你的机器是普通的服务器(commodity server),一个机器失败的概率是非常高的,这也就是Map/Reduce在处理海量数据情况下更适合的原因。在后面谈到LDA等Topic Model运算的时候,个人认为Map/Reduce不一定是最适合的,因为文档级的运算称不上是海量数据,最多是大量数据运算,它的量级是百万级左右,是否要用Map/Reduce的模型,值得探讨。但是用户级别的数据,是亿的量级,用Map/Reduce比较适合。
它是用调度计算代替调度,在处理数据时,是将程序复制到目标机器上,而不是拷贝数据到目标计算机器上。
计算流程非常类似于简单的Unix pipe,上次给出一个Unix pipe方式和Map/Reduce统计单词的功能的比较:
Pipe: cat input | grep | sort | uniq -c > output
M/R: Input | map | shuffle & sort | reduce | output
Hadoop支持多样的编程接口:
Java native map/reduce – 可以操作M/R各细节
Streaming – 利用标准输入输出模拟以上pipeline
Pig –只关注数据逻辑,无须考虑M/R实现
这里提供一个示例,帮助您理解M/R。假设输入域是 one small step for man0, one giant leap for mankind。在这个域上运行 Map 函数将得出以下的键/值对列表:
(one,1) (small,1) (step,1) (for,1) (man,1) (one,1) (giant,1)(leap,1) (for,1) (mankind,1)
如果对这个键/值对列表应用 Reduce 函数,将得到以下一组键/值对:
(one,2) (small,1) (step,1) (for,2) (man,1)(giant,1) (leap,1)(mankind,1)
结果是对输入域中的单词进行计数。
后面讨论的几种方法会用到机器学习算法,机器学习的算法有很多分类,其中有统计机器学习算法,统计机器学习在Hadoop上仅仅实现其逻辑是比较简单的。常用的统计模型可以大致归为下面两个类别:
指数族分布
大多工程上的实用分布都是指数族分布,指数族分布的形式比较简单:
它是参数和自变量的参数做内积,这个内积就是thetaTu(x)。指数族分布是由最大熵的原则推导出来的,即在最大熵的假设下,满足一定条件的分布可以证明出是指数族的分布。指数族分布函数包括:Gaussian multinomial, maximum entroy。
指数族分布在工程中大量使用是因为它有一个比较好的性质,这个性质可以批核为最大似然(Maximum likelihook, ML)估计可以通过充分统计量(sufficient statistics)链接到数据(注:摘自《Pattern Recognition》),要解模型的参数,即对theta做最大似然估计,实际上可以用充分统计量来解最大似然估计:
充分估计量大小是与模型的参数的空间复杂度成正比,和数据没有关系。换言之,在你的数据上计算出充分统计量后,最就可以将数据丢弃了,只用充分统计量。比如求高斯分布的均值和方差,只需要求出样本的与样本平方和。
指数族的混合分布
它不能通过计算得到充分统计量后将数据丢弃,但它在工程中使用的更多。比如有Mixture of Gaussians,Hidden Markov Model,Probability Latent Semantic Analysis(pLSI).
虽然概率模型有很多,但个人认为解决问题的思路却很相似。即如果问题本身可以描述成一种分布,比如某种指数族分布,那么就使用该种指数族分布。比如点击率的问题,它的最值总是0或1,那么它就可以描述成Bi-nomial分布,再比如明显的钟型分布,我们可以用高斯分布。如果分布本身比较复杂,无法用单一的指数族分布描述,我们可以将多个分布叠加来描述。基本上常用的技术就这两种。
Map的过程是收集充分统计量,充分统计量的形式是u(x),它是指数族函数变换函数的均值。所以我们只需要得到u(x)的累加值,对高斯分布来讲,即得到样本之和,和样本平方和。Reduce即根据最大似然公式解出theta。即在在mapper中仅仅生成比较紧凑的统计量, 其大小正比于模型参数量, 与数据量无关。在图中还有一个反馈的过程,是因为是EM算法,EM是模拟指数族分布解的过程,它本质上还是用u(x)解模型中的参数,但它不是充分统计量所以它解完之后还要把参数再带回去,进行迭代。我们前面提到的所有的模型,比如高斯分布,pLSI等等都可以用这个框架去解。
大家了解了这个框架后,可以先估计一个统计模型,然后先用这个框架试一下,如果它的确是这个统计模型,那么就用这个模型解就可以了。但如果数据并不满足服从该统计模型,那么我们只能回退到使用梯度族的方法去解,梯度族的方法也很简单,因为我们往往假设样本间有独立性,我们可以在每个样本上计算梯度,累加后得到总体的梯度,Map就是在每个样本上收集梯度,在Reduce上把它们加起来,然后按梯度下降,更新模型,再进行迭代。
下面有的内容是直接从PPT中拷贝的,没有什么解释,大家可以看一下A Universal Statistical Learning Platform on Hadoop Streaming这个视频。
Mapper
template <class TModel> class CTrainMapper : public CFeature, public IDataNnalyzer{ protected: TModel * pModel;
public: /// Comsume a data record \author Peng Liu virtual bool consume(const CRecord & record){ CFeature::consume(record); pModel -> accumulate(*this, 1.0f); return true; }
/// Produced statistics (or modified data in case needed) \author Peng Liu virtual bool produce(CRecord & record){ static bool first = true; pModel -> produce(record); if (record.getField("STAT") != NULL){ record.rmvField("PARAM") return true; } return false; }; }
|
Reducer
template <class TModel> class CTrainReducer : public IDataAnalyzer{ protected: TModel * pModel;
public: /// Comsume a data record \author Peng Liu virtual bool consume(const CRecord & record) {return pModel -> consume(record);}
/// Try to update model after all input data finish \author Peng Liu virtual void finish() {pModel -> update();}
/// Produced model \author Peng Liu virtual bool produce(CRecord & record){ pModel -> produce(record); if (record.getField("STAT") != NULL){ record.rmvField("STAT"); return true; } return false; } };
|
Map阶段所要计算的充分统计量为:
,Reduce阶段要计算的模型参数为:
void CGaussDiag::accumulate(CFeature & x, float occ) { size_t dim = getFeaDim(); assert(x.size() == dim); accumOcc(occ); for (size_t d = 0; d < dim; d ++) { stats[ d] += occ * x[d]; stats[dim + d] += occ * x[d] * x[d]; } } |
void CGaussDiag::update() { size_t dim = getFeaDim(); for (size_t d = 0; d < dim; d ++) { float X = stats[d], float X2 = stats[dim + d]; params[ d] = X / occ(); params[dim + d] = occ() / (X2 - X * X / occ()); } } |
先说明一下,个人感觉oozie并不好用。
Oozie的功能是连接多个Map/Reduce Job,完成复杂的数据处理的工作流引擎,它可以很方便定制数据流之间的依赖关系,一个Job可以依赖三种条件:数据,时间,其它Job。比如它可以指定一个Job在某些数据到达后开始,可以指定固定时间Job开始运行,还可以指定在其它Job运行完成后开始。
前面说它不好用是因为它的定义很麻烦,它使用hPDL(一种XML流程语言)来定义DAG(有向无环图)工作流:
<start to=‘ingestor‘/>
<action name=‘ingestor‘> … <ok to=“proc"/> <error to="fail"/> </action>
<fork name=“proc">
<path start=“proc1"/> <path start=" proc2"/> </fork>
<action name=‘proc1’> … <ok to="completed"/> <error to="fail"/> </action>
<action name=‘proc2‘> … <ok to="completed"/> <error to="fail"/> </action>
<join name="completed" to="end"/>
<kill name="fail"> <message>Java failed, error message… </kill>
<end name=‘end‘/>
并且oozie目前不支持Iteration,因为它是DAG。如果大家要用工作流引擎,可以推荐另一个工具是Linkin开发的Azkaban,它其实和Hadoop没什么关系,它是处理Linux环境下的Job的关系,所以它非常轻量级,也有一个比较方便的图形化界面去查看。Oozie是深度的与Hadoop结合,在通过API访问上,可能是有优势的,但简单的使用上又显得太重了。
介绍工作流引擎的原因是它在广告系统中是很重要的,因为广告系统的Hadoop上运行了很多任务,比如Audience Targeting,Click Model,Forecasting(流量预测),这些任务需要一个完善的调度机制,否则在运行环境中会非常混乱,数据的失败,Job的失败很难处理。工作流引擎会使整个系统更容易控制。
标签:
原文地址:http://www.cnblogs.com/94julia/p/4609928.html