文档章节

基于MaxCompute的图计算实践分享-Aggregator机制介绍

_夜枫
 _夜枫
发布于 2017/04/09 20:42
字数 2498
阅读 7
收藏 0

Aggregator是MaxCompute-GRAPH作业中常用的feature之一,特别是解决机器学习问题时。MaxCompute-GRAPH中Aggregator用于汇总并处理全局信息。本文将详细介绍的Aggregator的执行机制、相关API,并以Kmeans Clustering为例子说明Aggregator的具体用法。

Aggregator机制

如图1所示,Aggregator的逻辑分两部分,一部分在所有Worker上执行,即分布式执行,另一部分只在AggregatorOwner所在Worker上执行,即单点。其中在所有Worker上执行的操作包括创建初始值及局部聚合,然后将局部聚合结果发送给AggregatorOwner所在Worker上。AggregatorOwner所在Worker上聚合普通Worker发送过来的局部聚合对象,得到全局聚合结果,然后判断迭代是否结束。全局聚合的结果会在下一轮超步分发给所有Worker,供下一轮迭代使用。

aggregator
图1 Aggregator机制

Aggregator的API

Aggregator共提供了五个API供用户实现。下面逐个介绍5个API的调用时机及常规用途。

1. createStartupValue(context)

该API在所有Worker上执行一次,调用时机是所有超步开始之前,通常用以初始化AggregatorValue。在第0轮超步中,调用WorkerContext.getLastAggregatedValue() 或ComputeContext.getLastAggregatedValue()可以获取该API初始化的AggregatorValue对象。

2. createInitialValue(context)

该API在所有Worker上每轮超步开始时调用一次,用以初始化本轮迭代所用的AggregatorValue。通常操作是通过WorkerContext.getLastAggregatedValue() 得到上一轮迭代的结果,然后执行部分初始化操作。

3. aggregate(value, item)

该API同样在所有Worker上执行,与上述API不同的是,该API由用户显示调用ComputeContext#aggregate(item)来触发,而上述两个API,则由框架自动调用。该API用以执行局部聚合操作,其中第一个参数value是本Worker在该轮超步已经聚合的结果(初始值是createInitialValue返回的对象),第二个参数是用户代码调用ComputeContext#aggregate(item)传入的参数。该API中通常用item来更新value实现聚合。所有aggregate执行完后,得到的value就是该Worker的局部聚合结果,然后由框架发送给AggregatorOwner所在的Worker。

4. merge(value, partial)

该API执行于AggregatorOwner所在Worker,用以合并各Worker局部聚合的结果,达到全局聚合对象。与aggregate类似,value是已经聚合的结果,而partial待聚合的对象,同样用partial更新value。
假定有3个worker,分别是w0、w1、w2,其局部聚合结果是p0、p1、p2。假定发送到AggregatorOwner所在Worker的顺序为p1、p0、p2。那么merge执行次序为,首先执行merge(p1, p0),这样p1和p0就聚合为p1',然后执行merge(p1', p2),p1'和p2聚合为p1'',而p1''即为本轮超步全局聚合的结果。
从上述示例可以看出,当只有一个worker时,不需要执行merge方法,也就是说merge()不会被调用。

5. terminate(context, value)

当AggregatorOwner所在Worker执行完merge()后,框架会调用terminate(context, value)执行最后的处理。其中第二个参数value,即为merge()最后得到全局聚合,在该方法中可以对全局聚合继续修改。执行完terminate()后,框架会将全局聚合对象分发给所有Worker,供下一轮超步使用。
terminate()方法的一个特殊之处在于,如果返回true,则整个作业就结束迭代,否则继续执行。在机器学习场景中,通常判断收敛后返回true以结束作业。

Kmeans Clustering示例

下面以典型的KmeansClustering作为示例,来看下Aggregator具体用法。附件有完整代码,这里我们逐个部分解析代码。

1. GraphLoader部分

GraphLoader部分用以加载输入表,并转换为图的点或边。这里我们输入表的每行数据为一个样本,一个样本构造一个点,并用Vertex的value来存放样本。
我们首先定义一个Writable类KmeansValue作为Vertex的value类型。

  public static class KmeansValue implements Writable {

    DenseVector sample;

    public KmeansValue() { 
    }

    public KmeansValue(DenseVector v) {
      this.sample = v;
    }

    @Override
    public void write(DataOutput out) throws IOException {
      wirteForDenseVector(out, sample);

    }

    @Override
    public void readFields(DataInput in) throws IOException {
      sample = readFieldsForDenseVector(in);
    }
  }

KmeansValue中封装一个DenseVector对象来存放一个样本,这里DenseVector类型来自matrix-toolkits-java,而wirteForDenseVector()及readFieldsForDenseVector()用以实现序列化及反序列化,可参见附件中的完整代码。
我们自定义的KmeansReader代码如下:

  public static class KmeansReader extends 
    GraphLoader<LongWritable, KmeansValue, NullWritable, NullWritable> {

    @Override
    public void load(
        LongWritable recordNum,
        WritableRecord record,
        MutationContext<LongWritable, KmeansValue, NullWritable, NullWritable> context)
        throws IOException {
      KmeansVertex v = new KmeansVertex();
      v.setId(recordNum);

      int n = record.size();
      DenseVector dv = new DenseVector(n);
      for (int i = 0; i < n; i++) {
        dv.set(i, ((DoubleWritable)record.get(i)).get());
      }
      v.setValue(new KmeansValue(dv));

      context.addVertexRequest(v);
    }
  }

KmeansReader中,每读入一行数据(一个Record)创建一个点,这里用recordNum作为点的ID,将record内容转换成DenseVector对象并封装进VertexValue中。

2. Vertex部分

自定义的KmeansVertex代码如下。逻辑非常简单,每轮迭代要做的事情就是将自己维护的样本执行局部聚合。具体逻辑参见下面Aggregator的实现。

  public static class KmeansVertex extends
    Vertex<LongWritable, KmeansValue, NullWritable, NullWritable> {

    @Override
    public void compute(
        ComputeContext<LongWritable, KmeansValue, NullWritable, NullWritable> context,
        Iterable<NullWritable> messages) throws IOException {
      context.aggregate(getValue());
    }
  }

3. Aggregator部分

整个Kmeans的主要逻辑集中在Aggregator中。首先是自定义的KmeansAggrValue,用以维护要聚合及分发的内容。

  public static class KmeansAggrValue implements Writable {

    DenseMatrix centroids;
    DenseMatrix sums; // used to recalculate new centroids
    DenseVector counts; // used to recalculate new centroids

    @Override public void write(DataOutput out) throws IOException {
      wirteForDenseDenseMatrix(out, centroids);
      wirteForDenseDenseMatrix(out, sums);
      wirteForDenseVector(out, counts);
    }

    @Override public void readFields(DataInput in) throws IOException {
      centroids = readFieldsForDenseMatrix(in);
      sums = readFieldsForDenseMatrix(in);
      counts = readFieldsForDenseVector(in);
    }
  }

KmeansAggrValue中维护了三个对象,其中centroids是当前的K个中心点,假定样本是m维的话,centroids就是一个K*m的矩阵。sums是和centroids大小一样的矩阵,每个元素记录了到特定中心点最近的样本特定维之和,例如sums(i,j)是到第i个中心点最近的样本的第j维度之和。
counts是个K维的向量,记录到每个中心点距离最短的样本个数。sums和counts一起用以计算新的中心点,也是要聚合的主要内容。
接下来是自定义的Aggregator实现类KmeansAggregator,我们按照上述API的顺序逐个看其实现。
首先是createStartupValue().

  public static class KmeansAggregator extends Aggregator<KmeansAggrValue> {

    public KmeansAggrValue createStartupValue(WorkerContext context) throws IOException {
      KmeansAggrValue av = new KmeansAggrValue();

      byte[] centers = context.readCacheFile("centers");
      String lines[] = new String(centers).split("\n");

      int rows = lines.length;
      int cols = lines[0].split(",").length; // assumption rows >= 1 

      av.centroids = new DenseMatrix(rows, cols);
      av.sums = new DenseMatrix(rows, cols);
      av.sums.zero();
      av.counts = new DenseVector(rows);
      av.counts.zero();

      for (int i = 0; i < lines.length; i++) {
        String[] ss = lines[i].split(",");
        for (int j = 0; j < ss.length; j++) {
          av.centroids.set(i, j, Double.valueOf(ss[j]));
        }
      }
      return av;
    }

我们在该方法中初始化一个KmeansAggrValue对象,然后从资源文件centers中读取初始中心点,并赋值给centroids。而sums和counts初始化为0。
接来下是createInitialValue()的实现:

    @Override
    public KmeansAggrValue createInitialValue(WorkerContext context)
        throws IOException {
      KmeansAggrValue av = (KmeansAggrValue)context.getLastAggregatedValue(0);

      // reset for next iteration
      av.sums.zero();
      av.counts.zero();

      return av;
    }

该方法中,我们首先获取上一轮迭代的KmeansAggrValue,然后将sums和counts清零,其实是只保留了上一轮迭代出的centroids。
用以执行局部聚合的aggregate()实现如下:

    @Override
    public void aggregate(KmeansAggrValue value, Object item)
        throws IOException {
      DenseVector sample = ((KmeansValue)item).sample;

      // find the nearest centroid
      int min = findNearestCentroid(value.centroids, sample);

      // update sum and count
      for (int i = 0; i < sample.size(); i ++) {
        value.sums.add(min, i, sample.get(i));
      }
      value.counts.add(min, 1.0d);
    }

该方法中调用findNearestCentroid()(实现见附件)找到样本item欧拉距离最近的中心点索引,然后将其各个维度加到sums上,最后counts计数加1。
以上三个方法执行于所有worker上,实现局部聚合。接下来看下在AggregatorOwner所在Worker执行的全局聚合相关操作。
首先是merge的实现:

    @Override public void merge(KmeansAggrValue value, KmeansAggrValue partial) throws IOException {
      value.sums.add(partial.sums);
      value.counts.add(partial.counts);
    }

merge的实现逻辑很简单,就是把各个worker聚合出的sums和counts相加即可。
最后是terminate()的实现:

   @Override public boolean terminate(WorkerContext context, KmeansAggrValue value) throws IOException {
      // Calculate the new means to be the centroids (original sums)
      DenseMatrix newCentriods = calculateNewCentroids(value.sums, value.counts, value.centroids);

      // print old centroids and new centroids for debugging
      System.out.println("\nsuperstep: " + context.getSuperstep() + 
          "\nold centriod:\n" + value.centroids + " new centriod:\n" + newCentriods);

      boolean converged = isConverged(newCentriods, value.centroids, 0.05d);
      System.out.println("superstep: " + context.getSuperstep() + "/" 
          + (context.getMaxIteration() - 1) + " converged: " + converged);
      if (converged || context.getSuperstep() == context.getMaxIteration() - 1) {
        // converged or reach max iteration, output centriods
        for (int i = 0; i < newCentriods.numRows(); i++) {
          Writable[] centriod = new Writable[newCentriods.numColumns()];
          for (int j = 0; j < newCentriods.numColumns(); j++) {
            centriod[j] = new DoubleWritable(newCentriods.get(i, j));
          }
          context.write(centriod);
        }

        // true means to terminate iteration
        return true;
      }

      // update centriods
      value.centroids.set(newCentriods);
      // false means to continue iteration
      return false;
    }

teminate()中首先根据sums和counts调用calculateNewCentroids()求平均计算出新的中心点。然后调用isConverged()根据新老中心点欧拉距离判断是否已经收敛。如果收敛或迭代次数达到最大数,则将新的中心点输出并返回true,以结束迭代。否则更新中心点并返回false以继续迭代。其中calculateNewCentroids()和isConverged()的实现见附件。

4. main方法

main方法用以构造GraphJob,然后设置相应配置,并提交作业。代码如下:

  public static void main(String[] args) throws IOException {
    if (args.length < 2)
      printUsage();

    GraphJob job = new GraphJob();

    job.setGraphLoaderClass(KmeansReader.class);
    job.setRuntimePartitioning(false);
    job.setVertexClass(KmeansVertex.class);
    job.setAggregatorClass(KmeansAggregator.class);
    job.addInput(TableInfo.builder().tableName(args[0]).build());
    job.addOutput(TableInfo.builder().tableName(args[1]).build());

    // default max iteration is 30
    job.setMaxIteration(30);
    if (args.length >= 3)
      job.setMaxIteration(Integer.parseInt(args[2]));

    long start = System.currentTimeMillis();
    job.run();
    System.out.println("Job Finished in "
        + (System.currentTimeMillis() - start) / 1000.0 + " seconds");
  }

这里需要注意的是job.setRuntimePartitioning(false),设置为false后,各个worker加载的数据不再根据Partitioner重新分区,即谁加载的数据谁维护。

总结

本文介绍了MaxCompute-GRAPH中的Aggregator机制,API含义以及示例Kmeans Clustering。总的来说,Aggregator基本步骤是,
1)每个worker启动时执行createStartupValue用以创建AggregatorValue;
2)每轮迭代开始前,每个worker执行createInitialValue来初始化本轮的AggregatorValue;
3)一轮迭代中每个点通过context.aggregate()来执行aggregate()实现worker内的局部迭代;
4)每个Worker将局部迭代结果发送给AggregatorOwner所在的Worker;
5)AggregatorOwner所在worker执行多次merge,实现全局聚合;
6)AggregatorOwner所在Worker执行terminate用以对全局聚合结果做处理并决定是否结束迭代。

本文转载自:https://yq.aliyun.com/articles/69105

_夜枫
粉丝 10
博文 506
码字总数 0
作品 0
朝阳
后端工程师
私信 提问
阿里巴巴大数据计算平台MaxCompute(原名ODPS)全套攻略(持续更新20171127)

概况介绍 大数据计算服务(MaxCompute,原名ODPS,产品地址:https://www.aliyun.com/product/odps)是一种快速、完全托管的TB/PB级数据仓库解决方案。MaxCompute向用户提供了完善的数据导入方...

隐林
2017/05/05
0
0
MaxCompute 图计算用户手册(上)

概要 ODPS GRAPH是一套面向迭代的图计算处理框架。图计算作业使用图进行建模,图由点(Vertex)和边(Edge)组成,点和边包含权值(Value),ODPS GRAPH支持下述图编辑操作: 修改点或边的权值; ...

云花
04/08
0
0
【袋鼠云技术团队沙龙第13期】大数据计算服务MaxCompute线下Meetup-杭州站

活动简介: 大数据时代,计算的力量支撑着数据的价值。2018年双11 ,作为阿里巴巴统一的计算平台,MaxCompute单日数据处理超过500PB,平稳支撑电商混布单元在线流量洪峰12万笔/s交易。MaxCo...

墨羽MOYU
2018/11/21
3
0
【袋鼠云技术团队沙龙第13期】大数据计算服务MaxCompute线下Meetup-杭州站

活动简介: 大数据时代,计算的力量支撑着数据的价值。2018年双11 ,作为阿里巴巴统一的计算平台,MaxCompute单日数据处理超过500PB,平稳支撑电商混布单元在线流量洪峰12万笔/s交易。MaxCo...

墨羽MOYU
2018/11/21
42
0
所见不凡,带你探秘独角兽背后的大数据力量 | MaxCompute杭州Meetup报名中

阿里云MaxCompute X 众安保险 X 亲宝宝 X 中金易云 X 蚂蚁金服 邀您共赴杭州大数据计算线下Meetup 现在报名>> 双11的热度还未退却,大数据力量强势来袭。 是什么拖住了大数据开发者的后腿?集...

晋恒
2018/11/14
0
0

没有更多内容

加载失败,请刷新页面

加载更多

64.监控平台介绍 安装zabbix 忘记admin密码

19.1 Linux监控平台介绍 19.2 zabbix监控介绍 19.3/19.4/19.6 安装zabbix 19.5 忘记Admin密码如何做 19.1 Linux监控平台介绍: 常见开源监控软件 ~1.cacti、nagios、zabbix、smokeping、ope...

oschina130111
今天
9
0
当餐饮遇上大数据,嗯真香!

之前去开了一场会,主题是「餐饮领袖新零售峰会」。认真听完了餐饮前辈和新秀们的分享,觉得获益匪浅,把脑子里的核心纪要整理了一下,今天和大家做一个简单的分享,欢迎感兴趣的小伙伴一起交...

数澜科技
今天
7
0
DNS-over-HTTPS 的下一代是 DNS ON BLOCKCHAIN

本文作者:PETER LAI ,是 Diode 的区块链工程师。在进入软件开发领域之前,他主要是在做工商管理相关工作。Peter Lai 也是一位活跃的开源贡献者。目前,他正在与 Diode 团队一起开发基于区块...

红薯
今天
5
0
CC攻击带来的危害我们该如何防御?

随着网络的发展带给我们很多的便利,但是同时也带给我们一些网站安全问题,网络攻击就是常见的网站安全问题。其中作为站长最常见的就是CC攻击,CC攻击是网络攻击方式的一种,是一种比较常见的...

云漫网络Ruan
今天
11
0
实验分析性专业硕士提纲撰写要点

为什么您需要研究论文的提纲? 首先当您进行研究时,您需要聚集许多信息和想法,研究论文提纲可以较好地组织你的想法, 了解您研究资料的流畅度和程度。确保你写作时不会错过任何重要资料以此...

论文辅导员
今天
8
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部