Microsoft实习手记第二篇 Distributed Tensorflow

-
分布式机器学习目前主要有两种实现: Allreduce 和 参数服务器

差别主要在于参数的更新同步模式，Allreduce中各个机器的参数是同步更新的(synchronized)，参数服务器的话可以选择BSP ASP SSP各种同步模式[1]。 目前我所接触的工业界想要做的是用参数服务器做数据并行。

Tensorflow的API经常变更，现在也处于一个2.0在beta测试的时间节点。很多设计以及API都在一个更迭时期。这里记录一下怎么在一个分布式环境下用Tensorflow实现参数服务器，本文不会讨论对参数服务器的优化，背后原理，收敛证明等。（本文保质期会很短）


这视频是2017年TF官方对参数服务器的介绍，深入浅出，看了以后可以有一个大概的了解。


视频中提到了High Level API 和Low Level API 两种写法[2].
目前(2019/07/21)网络上大量的文档/博客讲的还是Low Level API的写法，让准备用TF写分布式机器学习的用户比较困惑。

先说Low Level API:
https://blog.csdn.net/u011026329/article/details/79190537
https://github.com/tensorflow/examples/blob/master/community/en/docs/deploy/distributed.md
这些链接给都是可运行的例子。

对于High Level API:
**这里**(很有参考价值)是TF官方给的Estimator的写法。它的Readme更新有点不及时，还是得对着代码来用。(写文的时候commit编号是448c31b6c688f93082ae5e51e9d5f0d1652b0ac1)
用High Level API主要是围绕这Estimator来完成自己的代码。Estimator在我们这里的好处在于迁移到分布式环境会很方便。
TF这里有一些封装好的Estimator，称为Premade Estimators。这里相当于给特定任务定义好了模型一些通用模型。如果用这种方式来完成自己的代码，那我们需要完成的主要就是对输入的处理：
  • 创建一个或多个输入函数。
  • 定义模型的特征列。
  • 实例化 Estimator，指定特征列和各种超参数。
  • 在 Estimator 对象上调用一个或多个方法，传递适当的输入函数作为数据的来源。
TensorFlow 把输入封装成一个流水线的形式丢给Estimator，并且把这个流水线视为 ETL 流程：
  1. 提取：从永久性存储（可以是 HDD 或 SSD 等本地存储或 GCS 或 HDFS 等远程存储）读取数据。
  2. 转换：使用 CPU 核心解析数据并对其执行预处理操作，例如图像解压缩、数据增强转换（例如随机裁剪、翻转和颜色失真）、重排和批处理。
  3. 加载：将转换后的数据加载到执行机器学习模型的加速器设备（例如，GPU 或 TPU）上。
这里给出一段样例代码来自Google IO18:

值得注意的是，这里TF让CPU处理数据的流水线，GPU来处理具体的机器学习运算。
并且还有一些针对分布式环境的优化方法，可以参考这个
这里记录一下当我们跑数据并行的时候，想要对数据进行切片然后分发到各个worker上，tf.data.Dataset 有shard 方法来做这个事情。

官方给出了以Iris Dataset为例怎么用premade estimator。这里的例子还是以estimator为中心，并没有focus在分布式的情况下（虽然改动不大）


对于使用自定义的Estimator，和上面的主要区别就是要自己定义模型函数。
def my_model_fn(
   features, # This is batch_features from input_fn
   labels,   # This is batch_labels from input_fn
   mode,     # An instance of tf.estimator.ModeKeys
   params):  # Additional configuration
classifier = tf.estimator.Estimator(
    model_fn=my_model,
    params={
        'feature_columns': my_feature_columns,
        # Two hidden layers of 10 nodes each.
        'hidden_units': [10, 10],
        # The model must choose between 3 classes.
        'n_classes': 3,
    })


这里介绍一些分布式TF的概念：

  • TF的分布策略
这个colab链接和这个文档虽然是官方对于TF分布式策略的文档，但总感觉说的很笼统，特别是参数服务器这块，并没有一个很完整的例子。目前TF有5种分布式策略：
tf.distribute.MirroredStrategy ：单机多卡，每个显卡都会保存一份完整模型，所有的参数同步更新。这里用的是Allreduce来做的参数同步，默认实现算法是Nvidia NCCL。也可以选择用其他allreduce实现。
tf.distribute.experimental.CentralStorageStrategy ：也是做的参数同步更新，但是参数存在CPU上，不会把模型存在GPU （Variables are assigned to local CPU or the only GPU. If there is more than one GPU, compute operations (other than variable update operations) will be replicated across all GPUs.）
tf.distribute.experimental.MultiWorkerMirroredStrategy ：这里相当于是第一种的多机多卡版本，一样是同步更新。
tf.distribute.experimental.TPUStrategy ：这种一样是类似于第一种，不过是跑在TPU上。
tf.distribute.experimental.ParameterServerStrategy ：终于到了我所需要的参数服务器部分。这里把集群里的机器分成worker和ps，这属于参数服务器架构的知识，这里不多说。特别的是，我们需要在每台机器的环境变量里定义好TF_CONFIG这个KV。下面具体讲了TF_CONFIG。

  • TF_CONFIG:
这是一个对于集群配置的定义，它包含了集群里的所有机器，以及他们的role：
cluster = {'master': ['master-ip:8000'],
           'ps': ['ps-ip1:8000','ps-ip2:8000'],
           'worker': ['worker-ip1:8000','worker-ip2:8000']}
TF_CONFIG = json.dumps(
  {'cluster': cluster,
   'task': {'type': worker, 'index': 0},
   'model_dir': 'gs:///',
   'environment': 'cloud'
  })
在用高阶API的情况下，在实例化Estimator的时候，需要传入Runconfig这个函数。它从环境变量'TF_CONFIG' 中读取关于集群配置信息以及当前机器对应role。除此以外还有一些其他配置信息比如分布式策略(train_distribute=...)等也在这里定义。具体可以看文档里的参数列表。

  • TFRecord：
通常工业界用分布式机器学习的场景是想做数据并行。这里TF能支持对于从非本地读取数据的方式，比如从HDFS读取[3]，从NFS读取(挂载到本地其实没区别)。下面是官方的说法：
  • 标准 POSIX 文件系统
  • HDFS - Hadoop 文件系统
  • GCS - Google Cloud Storage 文件系统
  • S3 - Amazon Simple Storage Service 文件系统
  • “内存映射文件”文件系统
如果读取一个远端的文件，其实网络开销是个很大的成本，包括对这个链接的建立。所以比起读多个零散的小文件（比如100个image），把他们打包成一个大小合适（100MB数量级）。Tensorflow的推荐做法是封装成一种叫TFRecord的数据格式。这里给了一个例子怎么转化现有数据集到TFRecord的形式。


Ref:
[1]: https://dl.acm.org/citation.cfm?id=2999748
[2]: https://www.tensorflow.org/guide/premade_estimators#the_programming_stack
[3]: https://medium.com/@matthewyeung/hadoop-file-system-with-tensorflow-dataset-api-13ce9aeaa107
TF官方指南(非文档): https://www.tensorflow.org/guide/premade_estimators
很实用的官方例子讲PS情况下用Estimator: https://github.com/tensorflow/models/tree/master/tutorials/image/cifar10_estimator
地点: Suzhou, Jiangsu, China

Comments

Post a Comment

Popular posts from this blog

Malware Report: iauzzy.exe

-
Image
Malware report of iauzzy.exe Summary This malware will create a file named ‘qqt.exe’ under the %WinDir%. This is the exact malicious executable file. qqt.exe is able to communicate to the ‘Command & Control’ server via IRC. The controller behind is able to execute any code; reboot the infected machine; clean the malware itself, etc.
Read more

Malware Report: withme.exe

-
Image
Malware Report withme.exe Summary This Withme.exe will drop a executable file ‘A-1980682553.exe’ to the location %AppData%. Also It will add itself to runonce registry. This is the exact malware. It’s able to communicate with C2 server, which domain is ‘sexwithme.info’ wia port 6667.
Read more

根因分析之iDice 文章复现

-
Image
1 Abstract  原文 给的场景是对于大规模集群的运维，当一条issue被上报的时候，通常会带上一些attribute，比如时间，软件版本，所处的国家，城市，具体在哪个机房，浏览器，操作系统等等。 在平时的时候，每天的上报的issue数量应该是差不多的，但有些时候，报错数量会突然上升，对于运维人员定位到问题所在往往很花时间。背后的问题可能是某个attribute或者某几个attributes的组合，文章称为effective combination。 这里通过三个步骤来从所有的attribute combination中找到最后可能引发报错激增的effective combination。 2 Challenge 想要找到所有effective combination的最暴力的做法就是，枚举所有attribute combination，然后以某种方式来衡量它对于总体issue激增的关系。 这里主要的问题在于搜索空间过大，本文最大的创新在于对搜索空间的剪枝。如果一共有m个attribute，那一共的组合方式就是(2^m - 1)种 3 Approach 文章把整个流程主要分成三个步骤（用做剪枝）和一个结果排序 Impact based Pruning: 基于影响的用户量做剪枝（不含时间信息） Change Detection based Pruning: 基于时间线上的突变来剪枝（时间序列处理） Isolation Power based Pruning: 计算一个类似信息熵增益的值来判断这个attribute combination的区分度 Result Ranking: 计算这个attribute combination对于全局的显著性 3.1 Impact based Pruning 对于一个effective combination而言，它必须影响足够多的用户。所以这里定一个threshold来过滤掉一些不可能的参数组合。（排除时间这个维度） 这里的目的是 寻找closed frequent itemsets，文章里的没有对算法进行详细描述，就说用BFS-based closed itemset mining algorithm。 在复现的时候，我先用了现成的找 frequent itemset 的算法（ Apriori /FPMax/ FPGrowt
Read more