横向联邦统计

Delta中的横向联邦统计在横向联邦任务框架下执行：

横向联邦任务框架

横向联邦任务框架对横向的隐私计算任务进行了抽象，所有的横向任务都可被拆分为多个计算单元，每个计算单元包括select, map, aggregate, reduce四个步骤。

用户编写的不同的横向计算任务，在真正执行前，都会先由Delta做一次转化，将整个计算过程转化为包含这四个步骤的多个单元，然后再发送到Delta Node进行执行。

本文详细讲解横向联邦统计的任务从用户编写的计算代码，到可被Delta Node执行的多个计算单元的转化过程。

用户如何定义横向联邦统计任务

首先，我们要解决的问题，就是用户如何编写代码，来实现他的横向联邦统计任务。关于这一点，当然是希望能最大限度地方便用户的编写。

这一点，我们从数据的角度来入手。也就是说，用户是在什么数据类型上进行统计计算的。对于统计任务来说，最常见的数据类型，就是各种表格了。 如果是一个单机的任务，用户要在表格数据上进行统计，那么他一般会使用pandas这个包来将数据读取成DataFrame，并通过DataFrame上的 各种算子，来对数据进行计算，最终得到他想要的结果。

那么，Delta的任务就很明确了，实现一套类似pandas.DataFrame的API，让用户可以像写pandas一样地来编写横向联邦统计任务。

提取任务的输入与输出

横向联邦统计任务

首先，我们就需要提取出用户编写任务的输入与输出。

对于任务的外部输入，也就是指客户端上的数据集，我们采用声明式的方式来让用户定义（参考Example里的dataset方法）。对于 其他内部的输入，比如说用户写在代码里的字面量，Delta可以自动地收集这些值，将他们作为任务的内部输入。

对于任务的输出，由于我们限定了，用户需要将所有的计算逻辑都写在一个方法里（参考Example里的execute方法），那么这个方法的 输出，就是整个任务的输出。任务执行时，我们只需要记录下execute方法输出的变量，作为任务的输出即可。

将单个算子转化为map与reduce

接下来，我们看如何将单个算子，转化为map与reduce。在这里，我们说的算子，指的就是pandas.DataFrame上的一些方法， 比如说sum，mean，std等等。

首先我们要分析，一个算子，需要转化为单纯的map操作，还是map+reduce操作，还是单纯的reduce操作。是的，你没看错， 根据算子参数的不同，同一个算子，可能转化为三种不同的操作。简单来说，当一个算子，需要联合多个不同客户端上的数据进行计算时， 就需要转化为map+reduce；否则的话，如果数据在客户端上，就转化为map，如果数据在服务端上，就转化为reduce。 举一个具体的例子，比如说mean这个算子，我们都知道它有一个axis参数，用来控制是沿着表格的行求和，还是沿着列求和。 在横向联邦的场景下，每个客户端持有数据的不同行，也就是说，每个客户端数据的行不相同，但是列相同。所以，在这种情况下， 当axis=0时（沿着行的方向，即对DataFrame的列求均值），mean算子需要转化为map+reduce；当axis=1时（沿着列的方向，即对DataFrame的行求均值）， 如果这个DataFrame在客户端上，那么mean就需要转化为一个map操作，如果在服务端上，就需要转化为一个reduce操作。 这里提一句，什么时候一个DataFrame，或者说数据会在服务端上，简单来说，一个map+reduce操作之后产生的结果，就位于服务端上， 当我们还需要对这个结果进行进一步的操作时，就会需要单纯的reduce操作。

接下来我们看如何将一个算子转化为map+reduce的操作。我们还是以mean为例。 对于单纯的map操作和单纯的reduce操作，它们都很简单，都是转化为pandas.DataFrame.mean就行了，比较复杂的是map+reduce操作。 我们先不考虑横向联邦的场景，就当做是一个传统的Map Reduce场景，看看要如何实现mean。 这点大家应该都能想到，只需要每个map操作输出本地数据的和sum以及数量count，reduce操作汇集所有map操作的输出之后求和， 得到所有数据的和sum'，以及所有数据的数量count'，最后计算sum' / count'即可。 带入到横向联邦的场景下来，在横向联邦中，安全聚合会自动将所有客户端map操作的输出求和，也就是说，reduce操作可以直接获得 sum'以及count'，不需要自己进行求和操作了。那么在横向联邦统计中，mean算子的map操作，依然输出本地数据的和sum以及数量count， 而reduce操作的输入，就直接是所有数据的和sum'以及所有数据的数量count'了，直接计算sum' / count'输出即可。

从mean算子这个例子出发，我们可以看出，map操作可能需要输出多个值，而其对应的reduce操作的输入，就是多个客户端上map操作的输出值之和。 所以，我们可以将map操作的输出组织成一个字典，字典的里的每个值，都需要是可加的；reduce操作的输入也是个字典，这个字典与map操作输出的字典 结构完全一致，只是字典的值变成了多个客户端输出的和。

将多个算子串联起来

在将单个算子转化为map与reduce之后，还不算完，我们还需要考虑如何将多个算子串联起来，毕竟用户编写的代码，不可能只包含一个算子。

在这里，Delta选择使用静态计算图的方式，来解决这个问题。如果用户用过Tenserflow 1.0，那应该很容易理解这个概念。 简单来说，用户编写的代码，并不是直接在Delta上执行的代码，用户编写的代码，只相当于定义了整个计算图，执行用户代码的结果， 是得到一张计算图。Delta会根据这张计算图，对它进行分析，将其转化为一系列的map与reduce操作，再插入相应的select和aggregate操作， 组成一个个Round，最后构建出整个横向联邦统计任务。

从实现上来说，用户在Delta中编写横向联邦统计的任务代码时，输入的数据结构delta.pandas.DataFrame，调用的是delta.pandas.DataFrame 上的各种算子。但实际上，delta.pandas.DataFrame并不包含数据，只是代表计算图上的一个数据节点。调用delta.pandas.DataFrame上的算子， 也只是在计算图上，添加了对应的map、reduce或者map+reduce节点。运行完用户编写的代码后，输出的数据节点，就是整个任务的输出了。 从输出节点出发进行遍历，就能得到整张计算图了。计算图中，那些没有入度的数据节点，自然就是任务的输入了。 得到计算图之后，我们需要根据计算图，来整理出任务中的每个Round。在每个Round中，都包含一个map和一个reduce操作，我们要做的，就是 将计算图上的map操作和reduce操作排好序，安排到每个Round的map和reduce中。这里需要指出的是，Round中的一个map和reduce，可以对应于计算图 中的多个map和reduce，只要这些计算图上的map或者reduce操作的输入数据在执行前已经准备完成，就可以把他们合并为一个map或者reduce操作。 构建好这些Round之后，整个任务的构建就完成了。之后我们需要做的，就是提交任务，开始执行了。

任务的执行

任务提交后，最后是在Delta Node的客户端与服务端上执行的。Delta Node在任务执行中，主要起两个作用： 一是与链上安全聚合系统交互，实现任务的创建、结算、以及安全聚合；二是为任务的执行提供数据。

首先看Delta Node的客户端。客户端除了监听任务创建、适时加入任务、下载任务外，在任务执行的过程中，主要执行 每个Round中的map操作。为了保证map操作能够顺利运行，需要根据map操作的输入，选择从本地加载相应的数据集、 从服务端下载所需的输入、从本地缓存中读取之前map操作的输出。map操作执行结束后，将属于map+reduce操作的输出通过链上安全聚合提交， 至于其他的输出，则会缓存在本地，供之后的map操作使用。

其次是Delta Node的服务端。服务端除了接受用户提交的任务，将任务提交到链上安全聚合系统外，在任务的执行过程中， 主要执行每个Round的reduce操作。reduce操作的输入，除了从链上安全聚合系统中获得外，还有可能有一些本地缓存的数据。 reduce操作执行结束后，得到的输出会缓存在本地，以供之后的reduce操作，或者其他客户端的map操作使用。

在每个Round开始执行时，选择加入这个Round的客户端，都会通过链上安全聚合系统，尝试去加入这个Round。 服务端也会通过链上安全聚合系统，对客户端进行选择。没有被选中的客户端，会直接退出这个Round的执行。

任务所有Round都结束后，客户端与服务端都会清理本地的缓存。同时，服务端还会根据任务的输出，将缓存中对应的值保存起来， 以供用户进行下载查看。