腾讯上万节点大规模集群的跨城自动迁移
作者简介：
方锦亮
腾讯 高级工程师
拥有多年大数据系统的平台运营和支撑平台建设经验，在海量设备管理和大数据系统运营上有深厚的技术积累。在腾讯主要负责 TDW （腾讯分布式数据仓库）的系统运营工作，主导 TDW 运营门户、发布中心、迁移平台等运营支撑平台的建设。
前言
作者在腾讯一直从事数据相关领域的系统运营和运营平台的建设工作。目前主要负责 TDW 的系统运营，TDW 是腾讯内部最大的离线处理平台，也是国内最大的 HADOOP 集群之一。
在运营这么大集群的时候，运营面临各种各样的难题，在解决这些难题的过程中，团队提炼出来的一个运营理念，用两句话去描述。
-
用建模的思路去解决运营的难题
运营的问题怎么解决？你必须用一些数据建模的办法，把这个难题解析清楚，然后我们再去考虑运营平台建设。 -
运营平台支撑模型运作
不是为了建设运营平台而建设，而是它必须有一定的运营理念。下文写到这样的运营理念是怎么贯穿在迁移平台的建设里面的。
本文主题主要包含以下几个方面：
介绍一下腾讯大规模集群 TDW，以及为什么做迁移。
我们的迁移模型是怎么样的。
迁移平台是怎么做的。
迁移策略。
1、腾讯大规模集群TDW
先介绍一下腾讯大规模集群，我们这里讲的集群是指 TDW。TDW 是腾讯分布式数据仓库，它是一个海量数据存储和计算平台。
为什么说是大规模集群？
记得我刚开始接手 TDW 运营的时候，很多年前，当时我们有400台的集群，觉得我们集群已经很大了，但是过了几年之后，我们要运营的集群已经达到4400台。
现在看来4400台还算挺大的，但是又过了几年后，我们规模到了8800台，这是我们的现状。我们的现状可以用三个指标来描述：
-
单集群 8800 台
-
支撑每天 20PB 扫描量
-
同时提供 200PB 存储能力
所以我们说是“大”，用苹果一句话说，biger than biger，我们预计到2017年底我们规模会达到2万台。
运营这么大规模的集群，运营人员自然会面临很多挑战。比如像设备运维、版本上线变更、配置管理、还有快速扩容等等。这些问题，我们都有相应系统去支撑，本文说的是我们遇到的另外一个头痛的问题：
集群不断膨胀，从400台到8800台，前期可以通过扩容解决。到目前这个阶段，8800台之后，我们发现扩容已经搞不定了。
为什么？
因为现有机房的容量和网络架构只能支撑这么大的规模，这时候我们需要将 TDW 迁移到其他城市更大容量的机房，这也就是我们面临的另一个问题，跨城迁移。
1.1 腾讯大数据处理平台整体架构
上文说了 TDW 的迁移原因，现在回过头来看 TDW 的整体架构。TDW 是腾讯大数据的处理平台的一部分，整个腾讯大数据处理平台包含了下面五层。
（腾讯大数据平台整体架构）
我们从下往上看：
-
最底层是数据存储层，包括 HDFS、HBase、Ceph、PGXZ；
-
第二层是资源调度层；
-
第三层是计算引擎层，包括 MR、Spark 和 GPU；
-
第四层是计算框架，包括 Hermes、Hive、Pig、SparkSQL、GraphX 等；
-
最上层是服务层，提供给外部数据分析能力和机器学习能力。
这是整个腾讯大数据平台，刚才说的 8800 覆盖了其中离线数据处理的部分。我们整个迁移覆盖了 HDFS、盖娅、MR、SPARK、HIVE、Pig 和 SparkSQL。
2、我们的迁移模型是怎样的？
2.1 跨城数据迁移到底难在哪里？
首先是运维工作量非常大。有上百P的数据要腾挪，有几十万任务需要切换，还有近万台的设备需要搬迁，这个事情对于运维来说工作量非常大。
其次，要保障业务无感知。迁移过程中系统要稳定可用，要保障数据不能丢失，不能把一份数据从一个地方搬迁到另外一个地方的时候，把数据弄丢了。
最后，要保障任务的计算结果准确而且任务的运行时长不能有明显的波动。
不管是运维工作量大还是业务无感知，这都还不是最致命的，对于跨城迁移来说，最致命的问题是：
当数据和计算分散在两个城市的时候，数据穿越可能造成专线阻塞，从而影响使用专线的所有系统，导致影响扩大化。
在介绍跨城迁移模型之前，我们先简单介绍两个方案，一个是双集群方案，一个是单集群方案。
方案一：双集群方案
双集群方案比较好理解，左侧跟右侧是两个城市的集群，双集群方案就是两套完全独立的系统，让它们独立去跑。
（双集群方案）
在说方案之前，我再深入介绍一下 TDW 里面的几个模块。我们只看左边就可以了。
左边从下面最底层是 GAIA，GAIA 负责资源调度。中间最左侧是数据采集 TDBank，它负责把各个业务线数据收集到 TDW。
TDW 的核心是计算引擎和存储引擎，存储引擎是放数据的地方，计算引擎提供 MapReduce 和 Spark 的计算能力。之上有查询引擎，最上面提供两个用户入口，任务统一调度和集成开发环境 IDE。
我举两个例子来说说各模块是怎么交互的。
案例一，数据是怎么进入 TDW 的？
首先业务数据经过数据采集模块，落地到存储引擎的某个目录下；统一任务调度 Lhotse 配置的一个入库任务，与 Hive 交互，将目录的数据转换成 Hive 表的数据。
案例二，数据是怎么计算的？
数据计算通过任务触发，任务是对数据的处理加工，比如统计日报的时候，计算任务对某个表做操作，把结果写回到另一个表中。
迁移是把存储和计算整套 TDW 平台，从一个城市搬迁到另外一个城市，双集群方案思路就很简单，在另外一个城市把所有系统都搭起来，跑起来就好了。
系统在两个城市之间是完全独立的，比如数据两份，计算两份，在这两个独立的系统之间不需要有任何的数据穿越（除了在迁移本身的数据穿越）。
这个方案最大优点就是不需要数据穿越，业务可以做到完全无影响，但是它最大缺点是需要大量的冗余设备。
方案二：单集群方案
下面讲一下单集群方案，它跟双集群差异点在哪里？
最核心的差别在于：存储不会同时在两个地方，要么在左边，要么在右边。
（单集群方案）
单集群方案有一个最大的优点就是不需要大量的设备，慢慢地把一部分设备，一部分业务，从左边迁移到右边。
这里会面临一个问题，比如刚才说到的一个计算的场景，如果没有控制好的话，会出现计算在左侧，数据已经跑到右侧去了，因为数据只有一份。任务跑起来的时候，左侧的计算引擎就会大量拉取右侧的数据，会对专线造成很大的风险。
对比一下刚才那两个方案，我们可以总结一下思路：在一个大的系统里，如果优先考虑成本，建议采用单集群方案。
单集群方案最大风险是跨城流量控制，跨城流量控制最重要的点是：数据在哪里，计算就去哪里，要不然就是穿越；如果访问的数据两边都有，哪边数据量大，计算就在哪边。
2.2 建立基于关系链的迁移模型
前面我们分析了一下我们实现跨城迁移的问题和方案，接下来我们为了解决跨城的流量控制降低跨城迁移的流量，我们引入一个基于关系链的迁移模型。
（一个关系链的例子）
我们需要知道数据流是怎么样来的，比如上面的一个关系链中，入库任务对最顶层的 HDFS 数据做一些加工处理，处理之后把结果保存到入库表；
分析人员基于这个入库表做各种计算和统计分析，比如统计某些指标，做关联性分析，这里配置了四个任务，这四个任务运行后产生新的结果表，其中还有两个结果表由下层的任务做进一步的处理，这样就产生了数据和任务的关系链。
引入关系链模型，它能帮助我们理清楚数据和任务的关系。
我们用椭圆描述数据，矩形描述对数据的加工，他们的连线表明访问数据的方向，是读还是写。这个关系可以用来指导我们的数据迁移，可以做到数据在哪里，计算就在哪里。
2.3 关系链的生成
接着的问题是在一个大的系统里关系链怎么生成？
在任务调度里面有一个概念，叫做依赖，用来描述任务的父子关系，父任务运行完成后子任务才允许运行。
原来我们没有做关系链的时候，这是纯粹的任务调度层的关系，虽然它有一定指导作用，但是不能直接应用在迁移里面，因为我们需要的是数据和任务的关系，而不仅是任务和任务的关系，我们需要从庞大的任务管理系统生成关系链，来指导数据迁移。
接下来介绍一个叫 hadoopdoctor 的运营工具，它是用做什么的？
它会把我们跑的任务信息采集回来，把它保存在 DB 里面，这些信息用于定位 MR 失败原因或性能分析。
它有一个主控模块，每五分钟去所有的 NodeManager 采集每个 MR 的配置和运行信息，比如说它的访问数据是什么，输出结果是什么。
为了支持迁移，我们改了一些逻辑，让 hadoopdocter 记录数据路径和任务ID，同时区分标识是读的还是写的。
把这个数据采集出来以后，我们就可以做关系链的分析。
（hadoopdoctor架构）
这里面采集到的路径会非常多，比如一个日报可能访问的是昨天某一个表的数据，比如访问量，就需要访问昨天的分区。采集出来的数据路径粒度非常细，它是包含日期的。
但是我们关注点并不需要到分区，我们关注的是表本身。所以我们把涉及到日期相关的路径规约掉，转成与日期无关的路径，数据规约对关系链分析有好处，归完之后减少了很多的数据量。
我们把最基础的信息采集到，它描述了一个任务，访问什么数据，产生什么数据。
经过我们的逻辑采集完之后，我们得到的是最原子的数据访问关系，就是一个任务对存储的操作，读或者是写，我们会产生非常多这样的原子关系，这种关系累计的结果就是关系链。
我们清洗出来一个最基础的关系，可以拼凑成一个大的关系链。
2.4 切分大关系链
关系链里面特别注意的地方，是一定要覆盖全面。
统计分析里不仅有日报，还有周报和月报，统计周期一定要覆盖较长的时间范围，这样才能把所有关系描述准确。
从最基础的原子关系聚合成关系链，可以用并查集算法。聚合出来的关系链，有大有小。对于小的关系链，很简单就可以把它迁走；但是我们发现一个很头痛的问题，聚合除了产生很多小的关系链，同时也产生了少量非常大的关系链，一个大的关系链可能包括超过十万个结点。
这时候我们发现回到原点，本来想把整个数据仓库从一个城市挪到另外一个城市，思路是将它打散生成多个关系链，最后也确实产生一些小的关系链，方便我们做迁移。但是遗留了一些大的关系链。
（一个大关系链的切分）
如果看这个图我们自然而然想：既然这么大的关系链迁不动，就先迁上面部分，这里就是把大的关系链切开的想法。
我们引入一些关键结点，这些关键点把大的关系链切分成多个小关系链。
那么哪个关系链结点才是最合适的？这个也很难找，大家可以想理论上是存在的，比如从图中标红色部分一刀切过去就可以一分为二，但是在上十万结点规模的关系链里是很难做到这个事情的。
2.5 引入HIVE双写表
把这个问题先留在后面，我们先做一个假设，已经找到合适的结点了，怎么实施关键结点的迁移，有两种思路：
-
一种是单份数据方案，比如把这份数据迁到另外一个城市，就让它穿越，很容易实现，但是不好控制流量；
-
另一种是双份数据方案，把关系链一分为二的时候，就把关键结点的数据变成两份。我们引入双写表的概念，双写表有两个 location。
（双写表和同步任务）
双写表的数据有两份，在城市A访问数据的时候，城市A 的 HIVE 会返到主的 location，城市B则返回备的 location，计算可以访问离自己近的存储。
这里，需要一个任务把城市A的数据同步到城市B中，这是一个同步任务，这个任务在数据在城市A生成后（总有个计算任务往里面写数据），把城市A的数据同步到城市B，保证两边的数据是一致的。
在迁移过程中把一个表升级成双写表的过程业务是无感知的，我们有机制保证数据双份，就近访问。
2.6 数据一致性保证
刚才说到双写表在城市A和城市B有两份数据，由同步任务负责同步，这时候我们会遇到一个问题，在城市A和城市B，它的下游任务会不会在同步任务没跑完之前就去访问这个数据？
（数据流转过程的任务依赖）
我们必须要保证数据的一致性，这通过增加对同步任务的依赖实现。
比如任务产生数据表1，下面三个任务会读取表1数据，这个依赖关系是上面这个任务是父任务，只有这个父任务跑完之后，下面三个子任务才能跑起来。这个依赖逻辑保证了三个子任务总能正确访问表-1的数据。
加入了同步任务之后，我们就保证了数据的一致性。比如这个图里，我们有两个任务从城市A迁到城市B，这时候我们要保持数据和计算的一致性，只需要保证城市B访问表-1数据的任务依赖同步任务即可。
2.7 最小化切分和关系链融合
回到一个大的关系链怎么拆分的问题，假设已经把它拆开了。
拆开的时候产生了很多小的关系链，把小的关系链从一个城市迁移到另外一个城市的时候，为了减少数据穿量引入双写表的概念，双写表加上任务依赖，保证了所有拆分出来的关系链有一个比较非常好的特性，就是不管产生多少个关系链，每个关系链都是随时可以迁移的。
还有一个问题：我们很难找到合适的关键点，对一个十万节点关系链，我们做了一些尝试，用遍历的方式查找所有可行的双写表，都不能把这么大的关系链拆开，我们发现不存在单个双写表可以拆开这么复杂的关系链。
大家可以想象一下，一个很复杂的图里面可能找不到一个结点能够将它切分成多个关系链。单个双写表做不到，则需要使用多个双写表，这时候找出合适双写表的算法就会非常复杂。
后来我们做了一个变通，让所有符合双写表规则的表都变成双写表，这样可以实现对一个大关系链的最小化拆分。之后我们对最小化拆分后的小关系链又做了融合，把很小的关系链并成规模适中的关系链。
这里面整体的思路，类似于工厂流水线打包，我们把关系链聚合，再把它拆分融合，变成一个个大小适中的关系链，就像工厂打包的一个个箱子，迁移就是把这些箱从从城市A运输到城市B的过程。
2.8 计算的迁移
我再补充一下计算的迁移，刚才上面架构里面，说计算需要从城市A迁移城市B，怎么切呢？
我们对每个任务加上城市 ID 的扩展参数，当这个任务需要从城市A切到城市B跑的时候，只需要改这个参数就好了。平台会记录城市 ID，并在不同系统上传递，进行准确的路由，这样实现任务是方便迁移的。
（任务迁移的路由）
我再总结一下迁移模型，我们面临运营上很头疼的事情，要做跨城迁移，我们解决的方案是基于关系链，把关系链打散，通过机制保证关系链的一致性，使得我们迁移能够跑起来。
3、跨城迁移平台的建立
刚才我们说的是模型，有一套模型指导我们做迁移这个事情，有了模型之后，需要一个平台来支撑这个模型。
关系从无到有，对最基础的关系进行聚合，形成关系链。找关系链的核心结点，把关系链拆分，将小的关系链融合成更大的关系链。迁移平台有一个模块专门负责关系链的这类操作。
另外一个模块是关系链的迁移模块，就是怎么把已经划分好的关系链从一个城市挪到另外一个城市，它涉及数据的迁移，任务的切换，普通表升级双写表，依赖任务和同步任务的处理。
此外还有一个模块是平台保障，数据校验，任务校验和跨城流量的监控。
3.1 关系链迁移模块
迁移模型解决了一件事情，就是从无到有，构造我们的关系链，然后把关系链从大拆小，拆到合适迁移一个个的规模适中的关系链。
关系链迁移是解决另外一个问题。一个关系链里包含任务和数据，在迁移过程中，它们的状态会变化。比如说数据还在写，TDW 的数据每天在不断变化。任务也可能还正在跑，没有结束。
也就是说，关系链并不是静止的状态，它是动态变化的。关系链迁移是要保障在动态变化的环境下，数据和任务的迁移是准确的。
（关系链迁移模块的状态变迁）
关系链迁移时，首先会看关系链里有什么双写表，先处理双写表，之后处理其他数据的迁移。数据迁移时，先按表分区展开，按分区迁移，这可以加快迁移的速度。
迁移开始前会通知用户，用户可以不用管，但是他需要被通知到。如果遇到什么异常，用户可以分析是不是因为这个变化导致。
通知完用户，会展开所有表的分区，对展开之后的分区做迁移操作，也就是 distcp。
当 distcp 到一定进度的时候，会做冻结任务的操作。关系链里面的任务可能有状态变化，冻结操作可以将其转变成不可变的状态。
把数据的写入任务冻结，写入任务可以通过关系链找到，如果没有关系链这种冻结操作也无法实施。迁移的过程不允许任务进行写入操作，因为写入操作会导致数据很难保证一致性。
把任务冻结之后进入等待数据一致的状态，在这个状态里还会不断对比两个城市之间的数据差异。
比如新增数据，数据不一样了就会再做一次数据同步的工作，直到数据完全一致，进入一致状态的时候，数据迁移的工作就完成了。之后进行任务迁移，任务迁移后就可以解冻任务，完成整个关系链的迁移。
这里最重要的是有一个冻结任务的逻辑，保证我们在数据迁移的时候，有一个时间段不会有任务在修改数据。冻结任务的时间越短越好，这就要求数据迁移的速度越快越好。
数据仓库里的表占用空间有大有小，对于大表，直接进行表级 distcp 的时候，通常起的 map 数很少，这时候并发上不来，速度也上不来。所以对于大表，需要采用分区级的 distcp。对于小表则相反，分区级的 distcp 往往是浪费。
另外关系链的迁移需要支持并发，关系链的在迁移过程消耗的资源不一样，有时候是网络流量大有时候网络流量小。关系链并发迁移，可以规避这种情况，实现接近满负载流量的迁移。
3.2 平台保障模块
平台保障模块，包括两大块的保障：一个是基础保障，还有是监控保障。
-
基础保障
做了两件事情，一个是数据校验，迁移完成两边数据要做校验，另外一件事情是任务的抽样重跑。我们引入一个思路，在关系链里面抽出一条垂直的路径，从根节点一路下来重做任务。一个关系链可能有几十个结点，抽样后只有四五个结点。对抽样出来的结点进行重跑，重跑后做数据比对，看看是不是一致的。这样可以保证数据是准确的，任务也是准确的。
-
监控保障
做了几件事情，一个是数据量的监控，迁移完成后，监控数据量的波动，看看数据量跟之前的是不是有明显波动。另外会监控所有迁移过的任务，看看它们在新的城市是否运行正常。最后，是流量的异常监控。数据和任务都验证成功，迁移也就是成功的了。数据已经从一个城市整体迁移到另外一个城市，任务也一起切换过来。最后要考虑的是，会不会有异常情况导致我们的跨城流量出现异常上涨。
我们有一个流量监控的机制，来解决一些例外情况导致的流量异常。通过加强流量的异常监控，并实现任务的自动切换。每五分钟采集正在运行的所有任务，以及它们访问的数据。
如果发现有任务在城市A的计算集群里面跑，但是它访问了城市B的数据，这种异常的情况会被监控起来。当数据穿越流量过高时，会自动把任务杀掉，同时自动进行任务切换。
4、跨城迁移平台的迁移策略
我们已经介绍了迁移模型和迁移平台，我们有一套模型去解决我们的运营面临跨城迁移的难点，然后我们也有一个平台支撑这个关系链迁移的逻辑。接下来我要再讲一下我们的迁移策略。
4.1 迁移集群独立部署
迁移集群需要独立部署，迁移最大的工作量是数据迁移，有很多数据要从一个地方同步到另外一个地方，数据迁移的方式就是做 distcp，distcp 是一个MR任务，MR 任务需要消耗计算资源。迁移集群最好独立开来，这样就不会影响正常的调度任务。
迁移集群最大的特点是网络流量会跑得很高，因为它只做一件事情，从源集群拉取数据，写到目标集群，观察迁移集群的网络流量的时候就会发现，它跑起来的时候，出入流量是一样的，它是网络流量高消耗的集群。
它有一个特点，它是低CPU消耗的集群。把它单独做为迁移集群，可以做到两个特性化配置，一个就是采用高网络配置的机型，比如使用万兆网卡的机型。另一个是采用低内存的任务配置以及对迁移集群的计算节点采用高并发的配置。
这样可以在尽量减少迁移集群的设备需求量的同时，大大提升迁移速度。我们采用40台机器的迁移集群就能支撑1P的迁移流量。
4.2 迁移的流量控制
（采用独立的迁移计算集群）
迁移的流量控制，这张图很有意思的地方是，当我们做数据迁移的时候，源集群（城市A HDFS集群），迁移计算集群，目标集群（城市B HDFS集群），这三个集群的流量有很有意思的关系。
源集群是一个大集群，通常有上千台机器；迁移计算集群则少得多，可能只有几十台。迁移计算集群在网络打满的情况下，对源集群的网络消耗可以计算出来，对目标集群的网络消耗也是可以计算出来的。
举个例子，如果迁移计算集群有40台的规模，这40台设备的网络用满的时候，源集群如果只有400台，网络也会被用满，网络用满的时候会对任务一定会有影响。
还有一个比较有意思的事情，目标集群的流量要比迁移集群流量大，原因是写数据时 Hadoop 有多份拷贝，会导致目标集群的流量翻倍增加。我们在使用迁移集群的时候，可以对迁移集群进行资源池管理，限定它的资源池大小，也就是限定迁移最大的并发数，从而对迁移流量进行控制。
4.3 集群的同步任务
再说说同步任务，同步任务对流量影响会比较小，因为同步任务方向和迁移方向是反方向的，迁移方向是城市A到城市B，同步任务是逆向回来的，所以流量很小。
我们尽量降低同步任务对业务的影响，建议使用独立的同步任务资源池，这个资源池可以更大一些，让同步任务快速完成，不影响其他任务。
4.4 HDFS集群的缩容扩容策略
最后，HDFS 集群的缩容扩容策略，在集群缩容的时候，要优先考虑集群整体下线，在缩容前要先进行数据清理和小文件合并。
另外，迁移的时候设备是一批批搬迁的，比如每一轮搬迁200台机器。这200台扩容到目标集群的时候，新扩容的节点在一段时间内不参与计算。因为 Hadoop 的 balance 机制会导致新扩容机器的网络流量被打满，直接影响计算任务的速度。
近期好文：
来GOPS · 深圳站 听听腾讯的大神们都在做什么？
就在4月21日-22日的GOPS2017深圳站哟。
GOPS2017·深圳站
我们都是资深的腾讯用户
有义务来GOPS关心一下腾讯的技术
-
会议地点：南山区圣淘沙酒店（翡翠店）
-
会议时间：2017年4月21日-22日
您可点击“阅读原文”，享受特惠折扣购票
发表评论