湖仓一体架构EMR元数据迁移DLF

概述

适用场景

• EMR元数据迁移至DLF。

• 元数据迁移验证。

• 数据一致性校验。

​

技术架构

本实践方案基于如下图所示的技术架构和主要流程编写操作步骤：

方案优势

• 统一元数据管理，快速构建数据库架构。

• 没有数据丢失风险。

• 较短的服务停机时间。

基础环境搭建

使用CADT创建资源

1. 登录CADT控制台，选择官方模板新建。

1. 由于元数据库使用了独立的RDS，需要先创建RDS再部署EMR，故先不部署EMR，先取消EMR资源部署。

1. 安全组增进开放3306端口（这里测试方便全部开放，建议仅针对VPC开放）。

1. 填写密码，保存应用。

1. 填写名称，保存应用。

1. 开始部署应用，等待部署完成。

1. 等待部署完成，创建数据库账号密码。

1. 创建Hive元数据库。

1. 在RDS控制台找到内网连接地址。

1. 部署完成的架构图把浏览模式变为编辑模式，开启EMR资源部署，设置EMR的登录密码，RDS数据库用户名和密码。

1. 再次部署资源。

1. 登录EMR master节点，使用独立RDS还需先初始化元数据库。

su hadoop
schematool -initSchema -dbType mysql

本方案使用的命令行文件可以从以下文件获取。点击下载文件。

一、上传测试数据

1. 使用scp命令拷贝测试数据到EMR集群。

scp 本地apache_logs.log文件path root@emr master公网ip:path

1. 登录EMR master节点，上传hdfs文件。

hdfs dfs -mkdir /user/hadoop/input
hdfs dfs -put /apache_logs.log /user/hadoop/input 

1. 使用以下命令查看hdfs文件，可见已成功上传。

hadoop fs -ls /user/hadoop/input 

1. 使用hive client创建数据库。

hive
create database log_data_warehouse;
use log_data_warehouse; 

1. 创建hive表。

CREATE TABLE apache_logs(
ipaddr STRING,
identity STRING,
username STRING,
accesstime STRING,
request STRING,
status STRING,
size STRING,
referer STRING,
agent STRING) 
PARTITIONED BY (year string,month string,day string)  
ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
"input.regex" = "([^ ]*) ([^ ]*) ([^ ]*) (\\[.*\\]) (\".*?\") (-|[0-9]*) (-|[0-9]*) (\".*?\") (\".*?\")",
"output.format.string" = "%1$s %2$s %3$s %4$s %5$s %6$s %7$s %8$s %9$s"
)
STORED AS TEXTFILE;

1. 给Hive表增加一个分区。

alter table apache_logs add partition (year='2015', month='05', day='17'); 

1. 把数据载入这个分区。

load data inpath '/user/hadoop/input/*'into table log_data_warehouse.apache_logs partition (year='2015',month='05',day='17');

1. 通过hive sql查看数据可以成功查询。

select * from log_data_warehouse.apache_logs where year='2015' and month='05' and day='17';

1. 可见log文件也被移动到当前分区下面了。

hadoop fs -ls -R /user/hive/warehouse/log_data_warehouse.db/apache_logs/year=2015/month=05/day=17

二、元数据迁移

1. 元数据迁移流程说明

请仔细阅读迁移流程，并严格按流程执行：

1. 建立DLF元数据迁移任务，并进行测试，确保迁移流程没问题。

2. 停止EMR服务，保证元数据不会变更。

3. 执行DLF元数据迁移任务。

4. 原hive元数据成功迁移DLF后，进行元数据校验。

5. 如果校验有问题，进行回滚（修改回原来hive metastore)

6. 元数据切换到DLF，重启EMR服务。

7. 校验相关任务执行情况是否正常。

如有问题：

1. 启动服务后，执行的任务不影响生产的，如第五步回滚。

2. 启动服务后，执行的任务可以重跑恢复的，如第五步回滚，并重跑任务。

3. 启动服务后，执行的任务无法重跑恢复的，联系阿里云支持同学。

2. 元数据迁移DLF

1. 进入数据湖构建（DLF）控制台，选择元数据迁移，创建迁移任务。

1. 选择MySQL数据库，类型为RDS（如果其他类型的数据库可以使用JDBC连接），配置相关的实例信息，进入下一步。

1. 输入名称，选择OSS的存储路径（如果没有OSS bucket请先到OSS控制台创建）。

1. 确认信息无误后提交。

1. 开始运行迁移任务。

1. 运行成功后，查看运行日志，可见迁移元数据起始时间（如果有错误参考日志进行排查）。

1. 迁移成功后，可以看到元数据库和表都自动创建成，确认元数据迁移成功。

3. 元数据校验

1. 在EMR master节点获取元数据校验工具包，点击下载。

su root
cd /
wget https://testbp-106.oss-cn-shanghai.aliyuncs.com/236/Democode/dlf-migrator-tools-1.0.0-RC1.jar

1. 创建YML配置文件，比如名称为migrator_config_validate.yml，内容如下。

!!com.aliyun.dlf.migrator.app.config.MigratorConfig
clientInfo:
  accessKeyId: $accessKeyId
  accessKeySecret: $accessKeySecret
  endPoint: dlf-vpc.$regionId.aliyuncs.com
  regionId: $regionId
  catalogId: $aliyunUid
mysql:
  connectionUri: jdbc:mysql://$host:$port/hivemeta
  driver: com.mysql.cj.jdbc.Driver
  userName: $userName
  password: $password
runOptions:
  batchSize: 200
  debug: false
  deleteBeforeCreate: false
  lowerCaseTableNames: true
  schema: hivemeta
  records: oss://$bucket/validate/
  objectTypes:
    - database
    - function
    - table
    - partition
  operations:
- validate
  locationMappings:
    - source: 原路径，如hdfs://user/hive/warehouse/
     target: 目标路径，如oss://dlf/warehouse/

配置项含义：

• DLF SDK Client相关信息。

• accessKeyId：用户阿里云主账号的Access Key Id。

• asscessKeySecret：用户阿里云主账号的Access Key Secret。

• endPoint：DLF的EndPoint，一般为 dlf-vpc.[region-id].aliyuncs.com。

• catalogId：用户阿里云主账号的UID。

• MySQL元数据库相关信息。

• connectionUrl：元数据库JDBC连接信息。

• driver：mysql driver，一般不需要改。

• userName：数据库用户名。

• password：数据库密码。

• RunOptions：运行时参数。

• batchSize：必选配置。批量调用DLF SDK的批次大小，最大不能超过500，太大可能会超时，太小效率慢，一般100左右可以。

• debug：必选配置。Spark是否会向console输出debug信息。

• deleteBeforeCreate：必选配置。如果在DLF创建对象的时候，如果对象已经存在，是否要删除后再重新创建。false的情况不会删除，冲突时会update；true的情况会删掉再创建；两者最大的区别是像创建时间之类的信息会有所不同。

• lownerCaseTableNames：必选配置。客户的元数据库里面的表是大写还是小写，目前需要客户自行判断。

• objectTypes: 必选配置。校验对象的类型列表，可以包含database、function、table、partition。

• operations：必选配置。校验工具执行的操作，指定validate进行元数据的比对校验，将把结果写到records路径上。

• records：必选配置。工具的运行结果日志，处理的记录，以及错误信息。

• locationMappings: 可选配置。对象location映射支持，应用于要修改替换database、table、partition等对象location信息的场景。有两个配置参数source和target，参数source表示要替换的路径（建议/结尾），target表示目标路径。

参考配置如下：

1. 上传为HDFS文件。

hdfs dfs -put /migrator_config_validate.yml /user/hadoop/

1. SSH到EMR集群，通过spark-submit命令提交运行校验任务（需要指定实际的校验工具包和配置文件的路径）。命令参考：

spark-submit \
 --master yarn \
 --deploy-mode client \
 --driver-memory 2G \
--executor-memory 2G \
--executor-cores 4 \
 --num-executors 3 \
--conf spark.sql.shuffle.partitions=200 \
 --conf spark.kryoserializer.buffer.max=128m \
 --conf spark.sql.autoBroadcastJoinThreshold=-1 \
--conf spark.sql.adaptive.enabled=false \
 --conf spark.sql.adaptive.join.enabled=false \
--conf spark.executor.heartbeatInterval=600s \
--conf spark.network.timeout=3600s \
 --class com.aliyun.dlf.migrator.app.MigratorApplication \
/dlf-migrator-tools-1.0.0-RC1.jar  \
/user/hadoop/migrator_config_validate.yml

1. 创建验证数据库。命令参考：

create table validate_database_result (leftResult string, rightResult string, status boolean, errorMessage string, fixResult string) stored as parquet location 'oss://testbp-106/validate/Databases';


create table validate_table_result (leftResult string, rightResult string, status boolean, errorMessage string, fixResult string) stored as parquet location 'oss://testbp-106/validate/Tables';


create table validate_function_result (leftResult string, rightResult string, status boolean, errorMessage string, fixResult string) stored as parquet location 'oss://testbp-106/validate/Functions';


create table validate_partition_result (leftResult string, rightResult string, status boolean, errorMessage string, fixResult string) stored as parquet location 'oss://testbp-106/validate/Partitions';

1. 通过以下命令验证是否存在不一致，为0表示全部一致。命令参考：

select count(*) from validate_database_result where status = false;
select count(*) from validate_table_result where status = false;
select count(*) from validate_function_result where status = false;
select count(*) from validate_partition_result where status = false;

1. 如果有数据不一致，通过以下命令查看详情，errorMessage字段表示不一致的具体情况。

select * from validate_database_result where status = false limit 10;
select * from validate_table_result where status = false limit 10;
select * from validate_function_result where status = false limit 10;
select * from validate_partition_result where status = false limit 10;

4. 元数据服务切换到DLF

1. 元数据校验没问题后，再进行元数据切换，进入EMR控制台，修改添加以下配置。

hive.imetastoreclient.factory.class=com.aliyun.datalake.metastore.hive2.DlfMetaStoreClientFactory
dlf.catalog.endpoint=dlf-vpc.cn-shanghai.aliyuncs.com
dlf.catalog.uid=156******8939
dlf.catalog.region=cn-shanghai
dlf.catalog.akMode=EMR_AUTO
dlf.catalog.sts.isNewMode=true
dlf.catalog.proxyMode=DLF_ONLY

METASTORE_ONLY, // read/write Metastore Only
METASTORE_DLF_FAILURE, // read/write Metastore first and write dlf allow failure
METASTORE_DLF_SUCCESS, // read/write metastore first and write dlf must success
DLF_METASTORE_SUCCESS, // read/write dlf first and write metastore must success
DLF_METASTORE_FAILURE, // read/write dlf first and write metastore allow failure
DLF_ONLY, // read/write dlf only

1. 重新启动HMS服务。

5. 验证元数据服务

1. 在EMR中使用hive命令创建一个新的元数据库。

1. 在DLF控制台查看新的元数据库已经创建成功，可见EMR集群的元数据库已经使用了DLF。

1. 验证数据查询成功（运行稳定后，可以切掉原来的RDS，仍可正常访问）。

use log_data_warehouse;
select * from log_data_warehouse.apache_logs where year='2015' and month='05' and day='17';