HDFS|HDFS跨集群的数据迁移 big|data|zookeeper|hadoop

文章目录

- 迁移之前需要先考虑的事：
- 迁移方案：
- 迁移工具distCp
- - 概述：
  - 原理
  - 使用方法：
- Map数目
- 不同HDFS版本间的拷贝
- Map/Reduce和副效应
- 迁移期间新老两个集群的资源消耗是怎样的
- 如何提高数据迁移速度
- 带宽如何限制
- 迁移之后的数据一致性如何校验
- 迁移之后的文件权限
- 迁移过程中老集群目录新增了文件，删除了文件怎么办
- 迁移中遇到文件已存在的情况怎么办？
- 迁移了一半，任务失败了怎么办？
- 遇到需要对一个文件增量同步怎么办？

迁移之前需要先考虑的事：

迁移总数据量
新老集群之间的最大带宽，在减少业务影响条件下最多可以使用多少带宽
在迁移过程中如何限制使用带宽
在迁移过程中，哪些文件可能发生删除、新增数据的情况。
迁移过程中，那些目录可能会发生新增文件的情况
迁移后的数据一致性校验
迁移后HDFS文件权限与老集群保持一致

迁移方案： 1.迁移数据量评估
通过hdfs dfs -du -h /命令查看各目录总数据量。按照业务划分，统计各业务数据的总量
2.指定迁移节奏
因为数据量大，而且带宽有限，所以在HDFS文件变化之前全部迁移是不可能的，因为HDFS中的文件会随着业务每天变化。所以应该按照业务、分目录、分批次迁移
3.迁移工具
Hadoop自带的数据迁移工具DistCp，可以通过简单命令完成数据迁移
hadoop distcp hdfs://nn1:8020/data hdfs://nn2:8020/
4.迁移时间
因为老集群仍然在使用，所以建议在老集群低负载运行的时间段进行迁移
5.新老集群带宽
询问运维新老集群之间的最大传输带宽，多少的带宽可以尽量少的影响业务
是否可以对新老集群之间的网络进行改造，例如通过接网线的方式提高网络带宽
6.数据迁移前状态评估
先尝试小数据量迁移，可以先进行100G-1T的数据迁移，以评估迁移时会遇到的问题
迁移工具distCp 概述：
DistCp(分布式拷贝)适用于大规模集群内部和集群之间拷贝的工具，它使用Map/Reduce实现文件分发，错误处理和恢复，以及报告生成。它把文件和目录的列表作为map任务的输入，每个任务会完成源列表中部分文件的拷贝。由于使用了Map/Reduce方法，这个工具在语义和执行上都会有特殊的地方。这篇文档会为常用DistCp操作提供指南并阐述它的工作模型。
原理
Distcp的本质是一个MapReduce任务，只有Map阶段，没有Reduce阶段，具备分布式执行的特性。在Map任务中从老集群读取数据，然后写入新集群，以此来完成数据迁移。

// org.apache.hadoop.tools.DistCp#main // org.apache.hadoop.tools.DistCp#run // org.apache.hadoop.tools.DistCp#execute /** * Implements the core-execution. Creates the file-list for copy, * and launches the Hadoop-job, to do the copy. * @return Job handle * @throws Exception */ public Job execute() throws Exception { Job job = createAndSubmitJob(); if (inputOptions.shouldBlock()) { waitForJobCompletion(job); } return job; }

/** * Create Job object for submitting it, with all the configuration * * @return Reference to job object. * @throws IOException - Exception if any */ private Job createJob() throws IOException { String jobName = "distcp"; String userChosenName = getConf().get(JobContext.JOB_NAME); if (userChosenName != null) jobName += ": " + userChosenName; Job job = Job.getInstance(getConf()); job.setJobName(jobName); job.setInputFormatClass(DistCpUtils.getStrategy(getConf(), inputOptions)); job.setJarByClass(CopyMapper.class); configureOutputFormat(job); job.setMapperClass(CopyMapper.class); // 无reduce阶段 job.setNumReduceTasks(0); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setOutputFormatClass(CopyOutputFormat.class); job.getConfiguration().set(JobContext.MAP_SPECULATIVE, "false"); job.getConfiguration().set(JobContext.NUM_MAPS, String.valueOf(inputOptions.getMaxMaps())); if (inputOptions.getSslConfigurationFile() != null) { setupSSLConfig(job); } inputOptions.appendToConf(job.getConfiguration()); return job; }

使用方法：
基本使用：
hadoop distcp hdfs://nn1:8020/foo/bar hdfs://nn2:8020/bar/foo
这个命令会把nn1集群的/foo/bar目录下的所有文件展开并存储到一个临时文件中，这些文件内容的拷贝工作被分配给多个map任务,然后每个TaskTracker分别执行从nn1到nn2的拷贝操作。
注意：注意DisCp使用绝对路径进行操作

在命令中可以指定多个源目录：
bash$ hadoop distcp hdfs://nn1:8020/foo/a hdfs://nn1:8020/foo/b hdfs://nn2:8020/bar/foo
或者也可以使用-f选项，从文件里获得多个源：
bash$ hadoop distcp -f hdfs://nn1:8020/srclist hdfs://nn2:8020/bar/foo
其中srclist的内容是：
hdfs://nn1:8020/foo/a
hdfs://nn1:8020/foo/b

当从多个源拷贝时，如果两个源冲突，DistCp会停止拷贝并提示报错信息，如果在目的位置发生冲突，会根据选项设置解决，默认会跳过已经存在的目标文件（比如不用源文件做替换操作）。每次操作结束时，都会报告跳过的文件数目，但是如果某些拷贝操作失败了，但在之后的尝试成功了，那么报告的信息可能不准确。
每个Tasktracker必须能够与源端和目的端文件系统进行访问和交互。对于HDFS来说，源和目的端要运行相同版本的协议或者使用向下兼容的协议。
拷贝完成后，建议生成源端和目的端文件的列表，并交叉检查，来确认拷贝的真正成功。因为DistCp使用Map/Reduce和文件系统API进行操作，所以这三者或它们之间有任何问题，都会影响拷贝操作。一些DistCp命令的成功执行可以通过再次执行带-update参数的该命令来完成，但用户在如此操作之前应该对该命令的语法很熟悉
值得注意的是，当另一个客户端同时在向源文件写入时，拷贝很有可能会失败。尝试覆盖HDFS上正在被写入的文件的操作也会失败。如果一个源文件在拷贝之前被移动或删除了，拷贝失败同时输出异常FileNotFoundException
选项：

标识····································	描述	备注
`-p[rbugp]`	Preserve： r:replication /b:blocksize/ u:user /g:group /p:permission/	修改次数不会被保留。并且当指定-update时，更新的状态不会被同步，除非文件大小不同（比如文件被重新创建）
`-i`	忽略失败	这个选项会比默认情况提供关于拷贝的更精确的统计，同时它还将保留失败拷贝操作的日志，这些日志还可以用于调试。最后，如果一个map失败了，但并没有完成所有分块任务的尝试，这不会导致整个作业的失败
`-log[logdir]`	记录日志到[logdir]	DistCp为每个文件的每次尝试拷贝操作都记录日志，并把日志作为map的输出。如果一个map失败了，当重新执行时这个日志不会被保留
`-m`	同时拷贝的最大数目	指定了拷贝数据时map的数目，但注意，并不是map数越多吞吐量越大
`-overwrite`	覆盖目标	如果一个map失败并且没有使用-i选项，不仅仅那些拷贝失败的文件，这个分块任务中的所有文件都会被重新拷贝，它会改变生成目标路径的语义，所以要小心使用这个选项
`-f`	使用作为源文件列表	这等价于把所有文件名列在命令行中。urilist_uri列表应该是完整合法的uri

更新和覆盖
这里给出一些-update和-overwrite的例子。考虑一个从/foo/a 和 /foo/b 到 /bar/foo的拷贝，源路径包括：

hdfs://nn1:8020/foo/a hdfs://nn1:8020/foo/a/aa hdfs://nn1:8020/foo/a/ab hdfs://nn1:8020/foo/b hdfs://nn1:8020/foo/b/ba hdfs://nn1:8020/foo/b/ab

如果没设置-update或-overwrite选项，那么两个源都会映射到目标端的/bar/foo/ab。如果设置了这两个选项，每个源目录的内容都会和目标目录的内容做比较。DistCp碰到这类冲突的情况会终止操作并退出
默认情况下，/bar/foo/a 和 /bar/foo/b目录都会被创建，所以并不会有冲突
现在考虑一个使用-update合法的操作：
distcp -update hdfs://nn1:8020/foo/a hdfs://nn1:8020/foo/b hdfs://nn2:8020/bar
其中源路径大小：

hdfs://nn1:8020/foo/a hdfs://nn1:8020/foo/a/aa 32 hdfs://nn1:8020/foo/a/ab 32 hdfs://nn1:8020/foo/b hdfs://nn1:8020/foo/b/ba 64 hdfs://nn1:8020/foo/b/bb 32

和目的路径/大小：

hdfs://nn2:8020/bar hdfs://nn2:8020/bar/aa 32 hdfs://nn2:8020/bar/ba 32 hdfs://nn2:8020/bar/bb 64

会产生：

hdfs://nn2:8020/bar hdfs://nn2:8020/bar/aa 32 hdfs://nn2:8020/bar/ab 32 hdfs://nn2:8020/bar/ba 64 hdfs://nn2:8020/bar/bb 32

只有nn2的aa文件没有被覆盖，如果指定了-overwrite选项，所有文件都会被覆盖
Map数目 DistCp会尝试着均分需要拷贝的内容，这样每个map拷贝差不多相等大小的内容，但因为文件是最小的拷贝粒度，所以配置增加同时拷贝（如map）的数目不一定会增加实际同时拷贝的数目以及总吞吐量
如果没使用-m选项，DistCp会尝试在调度工作时指定map的数目为min(total_bytes / bytes.per.map, 20*num_task_trackers)，其中bytes.per.map默认是256MB
建议对于长时间运行或定期运行的作业，根据源和目标集群大小、拷贝数量大小以及带宽调整map数目
不同HDFS版本间的拷贝对于不同Hadoop版本间的拷贝，用户应该使用HftpFileSystem。这是一个只读文件系统，所以DistCp必须运行在目标端集群上（更确切的说是能够写入目标集群的TaskTracker上）。源的格式是hftp:///（默认情况dfs.http.address是 :50070）。
Map/Reduce和副效应 map拷贝输入文件失败时，会带来一些副效应

除非使用了-i，任务产生的日志会被新的尝试替换掉
除非使用了-overwrite，文件被之前的map成功拷贝后当又一次执行拷贝时会被标记为“已忽略”
如果map失败了mapred.map.max.attempts次，剩下的map任务就会被终止，除非使用了-i选项
如果mapred.speculative.execution被设置为final和true，则拷贝的结果是未定义的

迁移期间新老两个集群的资源消耗是怎样的 DistCp是一个MapReduce任务，如果在新集群上执行就向新集群的Yarn申请资源，老集群只有数据读取和网络传输的消耗
如何提高数据迁移速度 DistCp提供了-m参数来设置map任务的最大数量（默认20），以提高并发性。注意这里要结合最大网络传输速率来设置
带宽如何限制 DistCp提供了-bandwidth参数来控制单个Map任务的最大带宽，单位是MB
限速原理：

// org.apache.hadoop.tools.util.ThrottledInputStream#read() public int read() throws IOException { // 每次从源HDFS读取数据的时候，会进行限速 throttle(); int data = https://www.it610.com/article/rawStream.read(); if (data != -1) { bytesRead++; } return data; } private void throttle() throws IOException { // getBytesPerSec()获取上次读取和此次读取这段时间内的速率（byte/s）与限速值作比较 while (getBytesPerSec()> maxBytesPerSec) { try { // 如果超速了则sleep一段时间 Thread.sleep(SLEEP_DURATION_MS); totalSleepTime += SLEEP_DURATION_MS; } catch (InterruptedException e) { throw new IOException("Thread aborted", e); } } }

迁移之后的数据一致性如何校验 DistCp负责进行CRC校验，可以通过-skipcrccheck参数来跳过校验来提高性能
迁移之后的文件权限 DistCp提供了-p参数来在新集群里保留状态（rbugpcaxt）（复制，块大小，用户，组，权限，校验和类型，ACL，XATTR，时间戳）。如果没有指定-p参数，则文件权限是执行MapReduce任务的用户权限，迁移完成后需要手动执行chown命令变更
迁移过程中老集群目录新增了文件，删除了文件怎么办应该尽量避免这种情况出现，DistCp在任务启动的时候就会将需要copy的文件列表从源HDFS读取出来，如果迁移期间新增了文件，新增的文件就会被漏掉。删除文件则会导致改文件copy失败，可以通过-i参数忽略失败
迁移中遇到文件已存在的情况怎么办？ Distcp提供了-overwrite 参数来覆盖已存在的文件。
迁移了一半，任务失败了怎么办？删除掉新集群中的脏数据，重新执行迁移命令。不加-overwrite参数，来跳过已存在的文件。
遇到需要对一个文件增量同步怎么办？ 【HDFS|HDFS跨集群的数据迁移】Distcp提供-append参数将源HDFS文件的数据新增进去而不是覆盖它。