开发者学堂课程【 SaaS 模式云数据仓库系列课程 —— 2021数仓必修课：Hadoop 数据如何同步至 MaxCompute 】学习笔记，与课程紧密联系，让用户快速学习知识。
课程地址：https://developer.aliyun.com/learning/course/55/detail/1038
Hadoop 数据如何同步至 MaxCompute
内容简介：
一、MMA 覆盖的主要场景
二、MMA 的主要功能
三、MMA 迁移服务架构
四、MMA Agent 技术架构和原理
五、环境准备
六、下载和编译工具包
七、MMA Agent 操作说明
八、进阶功能
九、进阶功能2：仅灵活的 hive 到 max compute 映射
十、单表/单分区迁移
十一、使用 Dataworks 自动迁移数据和工作流
十二、其他类型作业的迁移方案
十三、使用 MMA Agent 获得评估报告
一、MMA 覆盖的主要场景
二、MMA 的主要功能
迁移评估分析
数据迁移自动化
作业兼容性分析
对 Hadoop 平台进行诊断分析，评估数据迁移规模、作业迁移改造的数量、预估迁移后的成本，从而对迁移工作进行整体评估和决策。
对 Hive Meta 及数据进行检测扫描，自动在 MaxCompute 创建对应的 Meta，同时根据不同的网络环境，将 Hive 的数据自动转换并高吞吐地加载到MaxCompute 上，支持从 TB 级到 PB 级数据的迁移上云。
对 Hive 作业进行兼容性分析，识别出需要修改的任务并提供针对性的兼容性修改建议。对于用户自定义逻辑的分析任务，如UDF、MR/Spark作业等，我们将给出一般性的改造建议供用户参考。
对主流数据集成工具 Sqoop 进行作业的迁移转换，并自动创建 Dataworks 数据集成作业;支持主流 Pipeline 工 具，如 Oozie、Azkaban、Airflow 等自动迁移转化，并自动创建为 Dataworks 工作流及调度作业。
三、MMA 迁移服务架构
如下图，左边是客户 Hadoop 集群，右边是 Aliyun MaxCompute 即阿里云的大数据服务。 MMA 会跑在客户 Hadoop 集群上面，需要在客户的服务器上，这台机器上需要去部署 MMA 客户端工具，这个工具会帮助客户自动的去获取hivemeta 数据，还支持将 meta 信息自动转换成 MaxCompute ，批量创建表，之后会拉起数据同步的作业，调用 udf，udf 会 集成 SDK。
基于 MMA 客户端自动发现的 Meta 数据，基于 hiveMeta 数据会做一整个工作流的作业检查。
最后迁移过来以后对于 MMaxCompute 和 Dataworks 架构对接业务系统。
四、MMA Agent 技术架构和原理
MMA Agent 的工作流程主要分为四个步骤:
1、Metadata 抓取
2、MaxCompute DDL 与 Hive UDTF 生成
3、MaxCompute 表创建
4、Hive 数据
Meta Carrier 需要连接到 hivemete 里面，会自动将 meta 拉出来，在本地生成hivemeta 的结构。
Meta Processor 是基于第一个工具产出的结果，就是基于 hiveMeta 的数据转成DDL，即批量转成建表语句，包括数据类型的转换。
ODPS Console，基于Console可以将 Meta Processor 产出的 ODPS DDL 批量的通过 ODPS Console 去创建表。
基于 Data Carrier 去批量创建 HIVE UDTF SQL 作业。
五、环境准备
1、jdk 1.6+
2、Python 3+
3、Hive Client
4、能访问 Hive Server
5、网络连接 MaxCompute
场景举例：
·客户IDC->专线-> ECS+MaxCompute等，可以从ECS上访问
MaxCompute的endpoint ,但从IDC不可以
·需要增加 VBR 路由配置:参考
detail/57195.htmI?spm=a2c4g.11174283.6.579.33513a79ZnTEsX。
六、下载和编译工具包
1、下载源码︰切换到 odps-datacarrier-develop 分支，https/github.com/aliyun/aliyun-maxcompute-data-collectors?spm=a2c4g.11186623.2.8.422c4cO7MdjlpQ
2、解压下载的
aliyun-maxcompute-data-collectors-odps-datacarrier-develop.zip文件
在控制台运行 odps-data-carrier 目录下的 build.py 文件，编译生成 MMA 工具。
七、MMA Agent 操作说明
Bin 目录下有几个文件
Libs 下是所需要的 jar 包
[root@emr-header-1 mma_client_wz]# cd odps-data-carrier[root@emr-header-1 odps-data-carrier]# bin/meta-carrier -h
usage: meta-carrier -u <uri> -o <output dir [-h] [-d <database>][-t<table>] [--keyTab <path to keytab>] [--principalemetastore principal] [--system [key=value]+]
-d,--database <database> optional, specify a database
-h,--help optional, print help information
--keyTab <keyTab> optional, hive metastore's Kerberos keyTa
-o,--output-dir<output-dir> Required, output directory
--principal <principal>
Optional, hive metastore's Kerberosprincipal
--system system> system properties
-t,--table <table> 0ptional, specify a table
-u,--uri <uri>
Required, hive metastore thrift uri
[root@emr-header-1 odps-data-carrier]# bin/meta-carrier -u thrift：//127.0.0.1：9083-d dma demo -o meta
会生成一个 meta 目录
接下来会产生 partition meta 和 table meta。
接下来将 hivemeta 转换成 Max compute 的 dtl 语。
[root@emr-header-1 odps-data-carrier]# bin/meta-processor -h
usage: meta-processor -i <metadata directory -o coutput
directory>
-h,--help Print help information
-i,--input-dir <input-dir> Directory generated by meta carrier
-o,--output-dir <output-dir> output directory generated by meta
Processor.
-i 就是 input 输入目录
-o 就是 output 输出目录
Hive udtf sql 下 的 sql 是用于批量做数据迁移的时候写的 sql。
INFO mapred.FileInputFormat: Total input files to process : 1ds=20190925
ds=20190926ds=20190927ds=20190928ds=20190929
Time taken:_e.857 seconds,Fetched: 5 row(s)
optional arguments:-h, --help
show this help message and exit
--input INPUT
path to directory generated by meta processor
--odpscmd ODPSCMD path to odpscmd executable
[root@emr-header-1 odps-data-carrier]#
[rooteemr-header-1 odps-data-carrier]#
[root@emr-header-1 odps-data-carrier]# python36 bin/odps_ddl_runner.py -input output[root@emr-header-1 odps-data-carrier]# python36 bin/hive_udtf_sql_runner.py
Run hive UDTF SQL automatically.
optional arguments:
-h,--help
show this help message and exit
--input_all INPUT_ALL
path to directory generated by meta processor
--input_single_file INPUT_SINGLE_FILE，
path to a single sqtfile
--settings SETTINGS
path to extra settings to set before running a hive
--parallelism PARALLELISM
sql
max parallelism_of running hive sql
[rooteemr-header-1 odps-data-carrier]#python36 bin/hive_udtf_sql_runner.py -input_all
drwxr-xr-X 4 root root 4096 0ct 2919:23 dma demo-rw-r--r-- 1 root root 4525 0ct 2919:23 report.html[root@emr-header-1 output]#
[rooteemr-header-1 output]#
[root@emr-header-1 output]# sz report.html
1.使用 meta-carrier 采集 Hive Metadata
（1）解压工具包: odps-data-carrier.zip ，工具目录结构如下:
hive_udtf_sql_runner.py
meta-carrierl
meta-processor
network-measurement-tool（网络测量工具）
odps_ddl_runner.py（批量创建表）
proc_pool.py
_pycache_
proc_pool.cpython-36.pyc
sql-checker
extra_settings.ini
（2）获取 Hive metadata
（3）结果 hive metadata 输出的目录结构
说明：
① global.json 是一个全局的配置文件，包含了整个迁移过程中的一些配置
②每一个 database 会有一个独立的目录
③每一个表有一个以表名命名的 json 文件
④如果是分区表，还会有一个以表名为命名的 partition 的 json 文 件。
2.使用 network-measurement-tool
√测试 Hadoop 集群到 MaxCompute 各 Region 的网络连通质量
√测试网络上下行传输速率
Usage:
network-measure-tool --mode FIND |TEST
0ptions:
--—mode <mode>
-h ,--help
-p,--access-key <access-key>
0DPs access key, required inTEST mode
--project <project>
Print help information
0DPs project name，required inTEST mode
一t,--num-thread enum-thread>
Number of thread
--tunnel-endpoint <tunnel-endpoint> ODPS tunnel endpoint，optional-u,--access-id <access-id>
FIND (find available endpoints)
or TEST (test performance of asingle endpoint)
--endpoint cendpoint>
ODPS endpoint,required in TEST mode
ODPS access id,required in TEST
Mode
Example:
查找可用 endpoint:
odps-data-carrier/bin/network-measurement-tool --mode find
Output:
各个 endpoint 的连接情况(只输出可以连接的):
ENDPOINT: EXTERNAAL-BEI3ING: http://lservice.cn.maxconpute.aliyun.com/apiAVAILABILITY: true
ELAPSED TIME (ms): 4
ENDPOINT: EXTERONAL-HANGZHOU: http://service.cn.maxcompute.aliyun.com/apiAVAILABILITY: true
ELAPSED TINE (ms): 5
Example:
测试某个 endpoint 的读写性能:
odps-data-carrier/bin/network-measurement-tool --mode test l--endpoint <endpoint to test> \
-u<access id> -p <access key >\
--project=<project nane> --num-thread <number of thread>
3.使用 meta-processor 生成ODPS DDL和Hive UDTF SQL
（1）修改globle.json，自定义表、字段的生成规则
（2）生成 ODPS DDL 和 Hive UDTF SQL 了
4.使用 sql-checker 检查 Hive SQL 是否可以直接在 MaxCompute 执行
5.使用 odps_ddl_runner.py 批量创建表和分区
ODPS DDL 创建好以后，运行 odps_ddl_runner.py，将会遍历 meta-processor生成的目录，调用 odpscmd 自动创建 ODPS 表与分区
6.使用 hive_udtf_sql_runner.py 迁移数据
表和分区创建完成以后，运行 hive_udtf_sql_runner.py，将数据从 hive 上传至MaxCompute 。
hive_udtf_sql_runner.py 有两种模式，第一种将会遍历 meta-processor 生成的目录，调用 hive client 运行 hive udtf sql，从而将数据从 hive 上传至 ODPS 。第二种模式需要用户手动指定—个 hive sql 文件。
八、进阶功能
1、仅生成指定 database或table的metadata
在前面的 Demo 中，我们抓去了 hive 中所有 database 和表的
metadata ，但在很多环境下，我们倾向于一次处理一个 database 或一张表，因此meta-carrier 工具提供了抓取指定 database 或 table的 metadata 的能力:
执行:
sh odps-data-carrier/bin/meta-carrier -u
thrift://127.0.0.1:9083 -d test -t test -o meta
这里生成的 metadata 仅包含了 test.test 这张表。
九、进阶功能2：仅灵活的 hive 到 max compute 映射
在前面的 Demo 中，将 hive 的 test.test 表映射到 mc 中ODPS_DATA_CARRIER_TEST.test 这张表提供了更强大的能力，比如说修改 hive表到 mc 的表明与列名映射，设置 mc 中表的 life cycle， 增加 comment ，等等。
用户可以编辑 meta-carrier 生成的 metadata 来做到上述的事情。
十、单表/单分区迁移
在运行 hive sql 进行数据迁移的时候，提供了两种模式:
input_all 模式与 input_single_file 模式。
在 input_all 模式下，我们给一个 meta-processor 生成的目录，之后odps_hive_udtf_runner 会自动遍历该目录下的文件，并串行执行里面的hive sql，例如:
python3 odps-data-carrier/bin/hive_udtf_sql_runner.py --input_all processed/
在 input_single_file 模式下，我们给一个 hive sql 文件路径，odps_hive_udf_runner 会从该文件中读取 hive sql 并执行。例如∶
python3 odps-data-carrier/bin/hive_udtf_sql_runner.py--input_single_fileprocessed/test/hive_udtf_sql/single_partition/test_0.sql
input_single_file 模式可以帮助我们熟悉工具，并且在数据量大的场景下可以控制迁移的进度。
十一、使用 Dataworks 自动迁移数据和工作流
1、安装 MMA Agent 客户端工具:采集 Metadata& 生成 ODPS DDL
2、上传 Dataworks 项目描述文件
·根据模板(参见右图)生成 DataWorks 项目描述文档，打包为: dataworks_project.tgz 上传到 Dataworks。
·【注意】︰一期仅支持:1)打包文件手动上传;江2)支持OOIZE调度引擎的配置模板和Dataworks 工作流配置模板。
上传完成后， Dataworks 服务会根据 ODPS DDL 批量生成 MaxCompute 的table。
MaxCompute 的表创建完成后， Dataworks 服务会自动拉起 DataX 的数据同步任务，完成批量数据迁移。
3.项目描述文件（/project.xml）说明
tenantld: 用户在 dataworks 上的租户 ID;
name: 用户事先在 dataworks 上 创建好的项目空间名称;
owner :用户的阿里云账号 ID。
4.工作流描述文件 （/workflow.xml） 说明
十二、其他类型作业的迁移方案
1.UDF、MR 转移
支持相同逻辑的 UDF、MR 输入、输出参数的映射转换，但 UDF 和 MR 内部逻辑需要客户自己维护。
【注意】∶不支持在 UDF、MR 中直接访问文件系统、网络访问、外部数据源连接。
2.外表迁移
原则上全部迁到 MaxCompute 内部表。
如果必须通过外表访问外部文件，建议先将文件迁移到 OSS 或者 OTS，在MaxCompute 中创建外部表，实现对文件的访问。
【注意】: MaxCompute 外部表支持的格式包括 :ORC、PARQUET、SEQUENCEFILE、RCFILE.AVRO 和 TEXTFILE。
3.Spark 作业迁移
1.【作业无需访问 MaxCompute 表和 OSS 】用户 jar 包可直接运行，参照_《MaxCompute Spark 开发指南》第二节准备开发环境和修改配置。注意，对于spark 或 hadoop 的依赖必须设成 provided。
2.【作业需要访问 MaxCompute 表】参考_《 MaxCompute
Spark 开发指南》第三节编译 datasource 并安装到本地 maven 仓库，在 pom中添加依赖后重新打包即可。
【作业需要访问 OSS 】参考_《 MaxCompute Spark 开发指南》第四节在 pom中添加依赖后重新打包即可。
十三、使用 MMA Agent 获得评估报告
报告中将搬站风险分为两档,高风险 (HIGH RISK)与中等风险(MODERATE RISK)。·高风险意味着必须人工介入，例如出现了表名冲突， ODPS 完全不支持的类型等问题。
中等风险意味着迁移过程中可以自动处理，但是需要告知用户的潜在风险，例如Hive 数据类型到 ODPS 数据类型会带来的精度损失等问题。
