码迷,mamicode.com
首页 > 其他好文 > 详细

合约广告系统-Hadoop

时间:2015-06-30 12:26:50      阅读:199      评论:0      收藏:0      [点我收藏+]

标签:

 

 

Hadoop

Hadoop 概况

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包含两个部分:

1HDFS

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的副本放置策略是三个副本,一个放在本节点上,一个放在同一机架中的另一个节点上,还有一个副本放在另一个不同的机架中的一个节点上。

2MapReduce

有另一种分布式计算框架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上仅仅实现其逻辑是比较简单的。常用的统计模型可以大致归为下面两个类别:

指数族分布

大多工程上的实用分布都是指数族分布,指数族分布的形式比较简单:

技术分享

它是参数和自变量的参数做内积,这个内积就是theta­­­Tu(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/Reduce统计学习流程

 

技术分享

 

       Map的过程是收集充分统计量,充分统计量的形式是u(x),它是指数族函数变换函数的均值。所以我们只需要得到u(x)的累加值,对高斯分布来讲,即得到样本之和,和样本平方和。Reduce即根据最大似然公式解出theta。即在在mapper中仅仅生成比较紧凑的统计量, 其大小正比于模型参数量, 与数据量无关。在图中还有一个反馈的过程,是因为是EM算法,EM是模拟指数族分布解的过程,它本质上还是用u(x)解模型中的参数,但它不是充分统计量所以它解完之后还要把参数再带回去,进行迭代。我们前面提到的所有的模型,比如高斯分布,pLSI等等都可以用这个框架去解。

       大家了解了这个框架后,可以先估计一个统计模型,然后先用这个框架试一下,如果它的确是这个统计模型,那么就用这个模型解就可以了。但如果数据并不满足服从该统计模型,那么我们只能回退到使用梯度族的方法去解,梯度族的方法也很简单,因为我们往往假设样本间有独立性,我们可以在每个样本上计算梯度,累加后得到总体的梯度,Map就是在每个样本上收集梯度,在Reduce上把它们加起来,然后按梯度下降,更新模型,再进行迭代。

       下面有的内容是直接从PPT中拷贝的,没有什么解释,大家可以看一下A Universal Statistical Learning Platform on Hadoop Streaming这个视频。

Map/Reduce 基本统计模型训练

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;

    }

};

 

示例: Gaussian模型训练

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());

  }

}

 

Hadoop上的工作流引擎-oozie

       先说明一下,个人感觉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的失败很难处理。工作流引擎会使整个系统更容易控制。

 

 

合约广告系统-Hadoop

标签:

原文地址:http://www.cnblogs.com/94julia/p/4609928.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!