HDFS Distcp数据迁移与优化实践

1. 背景

对于HDFS集群而言，不可避免会将一个集群中的数据迁移到另外一个集群中。一般以下几种情况需要进行迁移：

1. hadoop2集群中的项目数据迁移到hadoop3中。
2. hadoop rbf的一个子集群block数量在2亿～3亿，需要将大项目迁移到其他空闲子集群。
3. 海外项目数据由于历史原因存放到国内集群，根据政策原因，需要迁移到海外。

在数据迁移时，可以使用HDFS提供的distcp工具进行迁移。

2. 迁移前期准备工作

2.1 迁移相关工作

对于数据迁移，并不只是简单地将一个集群中的HDFS数据迁移到另外一个集群中。还需要对周边组件进行适配：

1. 通过distcp工具将HDFS数据迁移到另外一个集群中。
2. metastore中项目db对应的所有表、分区location全部要改成新集群。
3. 作业提交的Yarn集群也会变更为新集群，需要调整对应队列大小。
4. 数据平台中，将所有直接访问的hdfs路径改为新集群。
5. 将Ranger权限迁移到新集群中。

2.2 准备工作

1. 先和数据拥有者沟通迁移事项，在拥有者同意后，确定迁移时间和迁移数据范围，制定对应的迁移计划表。并向所有数据使用者通知迁移事项，禁止在迁移过程中对项目目录进行写入操作。
2. 在新集群中增加router挂载信息，在yarn中增加用户队列，将用户队列调整至原集群大小。注意要根据子集群的block，挑选block数量最少的集群，避免过多的block对namenode造成压力。
3. 提前3天使用distcp工具将数据跨集群拷贝到新集群，这是为了正式迁移时，只会跨集群迁移增量数据。注意：迁移过程中需要过滤所有正在写数据的目录，如果distcp发现有新增文件，会直接报错。

3. 正式迁移流程

1. 停止项目下所有读写数据的离线作业和实时作业。
2. 将ranger权限迁移到新集群对应的实例中。
3. 将metastore的db、表、分区location改成新集群。
4. 释放原集群中该项目目录的租约，避免拿不到租约而导致迁移失败。
5. 使用hadoop distcp命令对数据进行迁移。
6. 恢复所有离线作业和实时写入。
7. 修改所有作业访问HDFS新集群。
8. 业务方验证作业读写正确性，一般业务会重跑最新一天数据。避免当天数据写入到旧集群，没有写入到新集群中。当天晚上询问重跑结果。如果有问题及时帮助用户解决。
9. 迁移第二天观察是否有数据写入到旧集群，将姿势调整写入到新集群中。如果没有写入到旧集群，数据移动到backup目录，对数据进行删除。

4. distcp迁移工具实践

通过一个最简单的例子了解distcp使用方法，它表示将cluster1中的/data/xxx数据迁移到cluster2集群的/data/xxx路径中：

hadoop distcp -pbugt -m 300 -update -delete hdfs://cluster1/data/xxx hdfs://cluster2/data/xxx

distcp工具迁移的基本思路如下：

1. distcp生成mapreduce作业，它会扫描要迁移的HDFS文件，按文件数平均分配给每个map。
2. 每个map负责将自己distcp的文件列表，依次进行跨集群拷贝。

distcp有较多重要参数，如下：

1. 保留权限 -p (preserve)：拷贝后的数据应该和拷贝前的数据一致，包括权限与元数据信息。推荐-pbugt参数。其他参数：r（副本数量）；b（块大小）；u（用户）；g（组）；p（权限）；c（checksum类型）；a（ACL）；x（xAttr）；t（时间戳）。如果只写-p，默认就是指保留所有权限，即-prbugpt。
2. 忽略错误 -i (ignore)：如果map在运行中出错，继续执行，不报错。这适合在源文件出现损坏时使用，避免作业终止。
3. 增量同步 -update：如果目标文件大小和源文件大小(fileSize)不一致，就替换为目的文件。
4. 删除 -delete：删除目的地址有，但是源空间没有的数据。
5. 全量同步 -overwrite：overwrite用于在目标集群如果存在相同的文件时进行覆盖。不管之前是否存在,都重新生成，适合全量同步。
6. 过滤 -filters：有些中间状态的文件没必要迁移，可以直接忽略掉。有些文件存在租约问题，在预迁移时，可以忽略掉，正式迁移时，停止写入，释放租约后再迁移。通过写fiters文件过滤，例如：.*/user/.*/\.Trash.* 可以过滤掉/user/*/.Trash目录，不对该目录进行迁移。
7. 带宽 -bandwidth：指定每个 map 的带宽，单位 MB/s。注意，如果一个节点运行N个map，每个map限制带宽是M，那么机器占用的带宽就是 N * M。
8. 直接写 -direct：直接写目的地址，避免通过写临时目录再rename，这样效率低。
9. 迁移列表的线程并发数 -numListstatusThreads 200：迁移前，构建迁移列表的并发数，可以设置200。
10. 设置迁移方式 -strategy dynamic：动态拷贝数据，运行较慢的map将会获迁移更少的数据，运行较快的map将会迁移更多的数据，避免集群中只剩几个map迟迟没有结束的窘境。
11. 跳过crc检查 -skipcrccheck true参数：如果文件大小相同，但是crc检查不同过时，可以指定-skipcrccheck为true跳过检查，避免作业退出。
12. map数量 -m 200：设置map数量时应该注意，全量迁移时，此时每个map基本上会占满带宽，如果此时有实时写入集群的情况，实时写入会因为网络问题导致写入延迟，kafka上数据会大量堆积。此时，应该将map数量尽量保持在DN数量的一半，保证集群半数DN可用，甚至应该限制带宽。增量迁移时，大部分时间只会校验数据，不会占用网络带宽，因此，map数量可以设置大一些。

JVM相关常用参数设置：

• -D "mapreduce.map.java.opts=-Xmx4096m -XX:PermSize=512m"
• -D "mapred.job.map.memory.mb=5120"
• -D "distcp.dynamic.max.chunks.tolerable=36000"
• -D "yarn.app.mapreduce.am.command-opts=-Xmx12288m"
• -D "yarn.app.mapreduce.am.resource.mb=8190"
• -D "yarn.scheduler.maximum-allocation-mb=12288"

生产环境中常用的distcp命令如下：

nohup hadoop distcp -D "distcp.dynamic.max.chunks.tolerable=36000" -D "yarn.app.mapreduce.am.command-opts=-Xmx65535m" -D "yarn.scheduler.maximum-allocation-mb=65535"   -D "yarn.app.mapreduce.am.resource.mb=30000" -D "yarn.app.mapreduce.am.resource.vcores=10" -pbug -strategy dynamic -numListstatusThreads 100 -direct -update -delete -bandwidth 2000000 -m 1000 hdfs://cluster1/user/yuliang02/tmp hdfs://cluster2/user/yuliang02/tmp 2>&1 >distcp_har.g17.log.tmp &

5. 基于Distcp snapshot迁移速度优化实践

5.1 背景

目前distcp默认情况会全量对比迁移前后的文件，所以迁移过程耗时很长。希望对这个情况进行优化。这种情况可以通过distcp snapshot解决，distcp比较两个snapshot的diff文件，直接拷贝这些diff文件，其他文件直接忽略不放入迁移比较过程中，避免迁移时花费大量时间进行数据对比。经过实践，原本6h的迁移时间可以降低到20min。

5.2 distcp snapshot迁移流程

以cluster1集群迁移到cluster2为例。迁移步骤如下：

1. 首先，对于第一次开始数据备份的文件，先在cluster1创建一个快照；利用DistCp拷贝全量的快照数据到cluster2集群；
2. 后续的每次预迁移都会生成一个快照，和前一天的快照基于distcp–differ参数进行对比，将有更新的部分再同步到cluster2集群；
3. 同步完成后，会删除前一天的快照，这样就完成了每日数据的增量同步。
4. 正式迁移时，停止离线和实时作业，新建snapshot，进行迁移即可。
5. 修改hive元数据。
6. 迁移完成后，删除所有snapshot，启动离线和实时作业。

命令流程如下：

首先cluster1创建快照s1:

# 创建快照
hdfs dfsadmin -allowSnapshot hdfs://cluster1/user/yuliang02/
hdfs dfs -createSnapshot hdfs://cluster1/user/yuliang02/ s1

将cluster1 s1快照复制到cluster2中，复制完后，cluster2创建快照s1。此时cluster1和cluster2中的

s1快照内容一模一样：

# cluster1迁移快照
nohup hadoop distcp -pbugt -m 100 -bandwidth 2000000 -strategy dynamic -direct -overwrite hdfs://cluster1/user/yuliang02/.snapshot/s1 hdfs://cluster2/user/yuliang02 > tmp/2.log 2>&1 &

# 创建cluster2快照
hdfs dfsadmin -allowSnapshot hdfs://cluster2/user/yuliang02/
hdfs dfs -createSnapshot hdfs://cluster2/user/yuliang02/ s1

正式迁移时，停止所有离线和实时的读写作业。在cluster1中创建s2快照，并比较cluster1的s2快照和cluster1的s1快照差异，将差异的文件进行迁移：

# 创建cluster1的最新快照
hdfs dfs -createSnapshot hdfs://cluster1/user/yuliang02/ s2
hadoop distcp -pbugt -m 100 -bandwidth 2000000 -strategy dynamic -direct  -update -diff  s1 s2 hdfs://cluster1/user/yuliang02/ hdfs://cluster2/user/yuliang02/

最终，删除所有快照：

# 删除cluster1的snapshot
hdfs dfs -deleteSnapshot hdfs://cluster1/user/yuliang02/ s1
hdfs dfs -deleteSnapshot hdfs://cluster1/user/yuliang02/ s2

# 删除cluster2的snapshot
hdfs dfs -deleteSnapshot hdfs://cluster2/user/yuliang02/ s1

# 禁用cluster1和cluster2的snapshot
hdfs dfsadmin -disallowSnapshot hdfs://cluster1/user/yuliang02
hdfs dfsadmin -disallowSnapshot hdfs://cluster2/user/yuliang02

5.3 hive元数据变更流程

hive使用的是mysql存储元数据，因此，需要将mysql中元数据路径从cluster1改成cluster2。包含db路径、table路径、分区路径、serde中的路径：

#db路径改成cluster2
sql1="update DBS set DB_LOCATION_URI = REPLACE(DB_LOCATION_URI, 'hdfs://cluster1/', 'hdfs://cluster2/') where NAME='${us_proj}';select ROW_COUNT();";

#table路径改成cluster2
sql2="update SDS SET LOCATION = REPLACE(LOCATION,'hdfs://cluster1/','hdfs://cluster2/') where SD_ID in (select SD_ID_middle from (select
    c.SD_ID as SD_ID_middle
from
    DBS a
          left join TBLS b on a.DB_ID=b.DB_ID
          left join SDS c on b.SD_ID=c.SD_ID
          left join SERDES d on c.SERDE_ID=d.SERDE_ID
          where a.NAME='${db}' and c.LOCATION  like '%hdfs://cluster1/%' ) as middle );select ROW_COUNT();"

#分区路径改成cluster2
sql3="update SDS SET LOCATION = REPLACE(LOCATION,'hdfs://cluster1/','hdfs://cluster2/') where SD_ID in (select SD_ID_middle from (select
    f.SD_ID as SD_ID_middle
from
    DBS a
          left join TBLS b on a.DB_ID=b.DB_ID
          left join SDS c on b.SD_ID=c.SD_ID
          left join SERDES d on c.SERDE_ID=d.SERDE_ID
          left join PARTITIONS e on b.TBL_ID = e.TBL_ID
          left join SDS f on e.SD_ID=f.SD_ID
          left join SERDES g on f.SERDE_ID=g.SERDE_ID
          where a.NAME='${db}' and f.LOCATION  like '%hdfs://cluster1/%') as middle );select ROW_COUNT();"

#serde路径改成cluster2
sql4="update SERDE_PARAMS SET PARAM_VALUE = REPLACE(PARAM_VALUE,'hdfs://cluster1/','hdfs://cluster2/') where SERDE_ID in  (select SD_ID_middle from (select
i.SERDE_ID as SD_ID_middle
from
DBS a
  left join TBLS b on a.DB_ID=b.DB_ID
  left join SDS c on b.SD_ID=c.SD_ID
  left join SERDES d on c.SERDE_ID=d.SERDE_ID
  left join SERDE_PARAMS i on i.SERDE_ID=d.SERDE_ID
  where a.NAME='${db}' and i.PARAM_VALUE  like '%hdfs://cluster1/%') as middle);select ROW_COUNT();"