开发者社区> 隐林> 正文
阿里云
为了无法计算的价值
打开APP
阿里云APP内打开

Hadoop数据迁到MaxCompute

简介:
+关注继续查看

通过最佳实践帮助您实现上述案例效果

Step1:数据准备

接下来,我们需要准备好一张表及数据集;

  • Hive表名:hive_dplus_good_sale;
  • 是否分区表:分区表,分区名为pt;
  • hdfs文件数据列分隔符:英文逗号;
  • 表数据量:100条。

源hive表建表语句

CREATE TABLE IF NOT EXISTS hive_dplus_good_sale(
   create_time timestamp,
   good_cate STRING,
   brand STRING,
   buyer_id STRING,
   trans_num BIGINT,
   trans_amount DOUBLE,
   click_cnt BIGINT,
   addcart_cnt BIGINT,
   collect_cnt BIGINT
   )
   PARTITIONED BY (pt string) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' lines terminated by '\n' ;

数据预览如下:

TB1eZQQOVXXXXcDXFXXXXXXXXXX-934-193.png

TB1j_NePXXXXXXDXpXXXXXXXXXX-1048-422.png
Hdfs&datax类型对照表

在Hive表中的数据类型 DataX 内部类型
TINYINT,SMALLINT,INT,BIGINT Long
FLOAT,DOUBLE,DECIMAL Double
String,CHAR,VARCHAR String
BOOLEAN Boolean
Date,TIMESTAMP Date
Binary Binary

Step2:MaxCompute目标表准备

1.1 开通MaxCompute
阿里云实名认证账号访问https://www.aliyun.com/product/odps ,开通MaxCompute,选择按量付费进行购买。
TB1IKEAOVXXXXcRaXXXXXXXXXXX-1158-487.png
TB1hTApOVXXXXbWaFXXXXXXXXXX-1243-351.png
TB1IHkOOVXXXXa5XVXXXXXXXXXX-1208-337.png

1.2 数加上创建MaxCompute project

操作步骤:

步骤1:进入数加管理控制台,前面开通MaxCompute成功页面,点击管理控制台,或者导航产品->大数据(数加)->MaxCompoute(https://www.aliyun.com/product/odps) 点击管理控制台。

TB1SNkTOVXXXXbiXFXXXXXXXXXX-1281-473.png

步骤2:创建项目。付费模式选择I/O后付费,输入项目名称:

TB1BLoEOVXXXXaRapXXXXXXXXXX-1361-506.png

步骤3:创建MaxCompute表。进入大数据开发套件的数据开发页面:

TB1ZFhePXXXXXcDXXXXXXXXXXXX-1023-211.png

新建表

TB18IACOVXXXXb_XVXXXXXXXXXX-1044-336.png

对应建表语句:

CREATE TABLE IF NOT EXISTS mc_dplus_good_sale (
   create_time DATETIME COMMENT '时间',
   good_cate STRING COMMENT '商品种类',
   brand STRING COMMENT '品牌',
   buyer_id STRING COMMENT '用户id',
   trans_num BIGINT COMMENT '交易量',
   trans_amount DOUBLE COMMENT '金额',
   click_cnt BIGINT COMMENT '点击次数',
   addcart_cnt BIGINT COMMENT '加入购物车次数',
   collect_cnt BIGINT COMMENT '加入收藏夹次数'
)
PARTITIONED BY (pt STRING);

Step3:通过Datax进行数据迁移

MaxCompute中的结构化数据是以表的形式存在,我们可以通过工具或者API将数据直接写入表中。

网络传输

  • datax: https://github.com/alibaba/DataX 本case我们选用datax方式进行数据迁移。
    datax是阿里开源的一款工具,可以适配常见的数据源,包括关系数据库、文本文件等,这是一款单机的软件,适用于中小数据量的传输,传输效率按照单机网卡 50%(约50MB/s)计算,一天可以传输的数据量为

50MB/s 60 60 * 24 / 1024/1024 ≈ 4T
具体的传输效率会取决于网络环境(机房出口带宽、本地网络),机器性能,文件格式与数量等多个因素。
使用datax传输数据到ODPS的部署结构如下,在ECS机器或一台物理机器上部署好dataX,此机器要能够同时连通数据源与MaxCompute(原ODPS)服务。

TB108QROVXXXXcNXFXXXXXXXXXX-1290-128.png

磁盘导入

当数据量比较大(PB级别以上)时,可以考虑通过磁盘将数据导入ODPS,需要磁盘导入请联系相应的大数据业务架构师。

API: Tunnel SDK

当数据源较特殊,无法使用以上现有的工具的时候,可以用SDK编写适配的工具,从数据源读出并写入ODPS中,Tunnel是数据上传的底层接口,可以同时接受大量客户端并行的写入。
https://help.aliyun.com/document_detail/27837.html?spm=5176.doc34614.6.144.RyeirI

ECS机器部署

前提准备:获取ECS的主机名和IP:

TB16pI2OVXXXXbkXpXXXXXXXXXX-972-297.png

操作步骤:

步骤1:增加调度资源。大数据开发套件->项目管理->调度资源管理->增加调度资源,输入名称和标识hive2mc:
TB12kdlPXXXXXXnXXXXXXXXXXXX-1352-292.png

步骤2:添加配置ECS服务器:
TB1kdMTOVXXXXXDaXXXXXXXXXXX-1216-375.png
输入前提准备中获取到的ECS服务器名称和IP,点击添加。

步骤3:初始化ECS服务器:
TB1GnozOVXXXXcxapXXXXXXXXXX-1200-318.png

按照弹框提示,登录ECS服务器(登录外网IP)在root用户下执行两个命令,即部署相关datax等服务。

命令执行成功后,回到“管理服务器”页面,点击刷新按钮可以看到服务器状态为“正常”状态:
TB19nhaPXXXXXXFXXXXXXXXXXXX-1090-152.png

此时该项目的调度资源组就有两个:默认资源组、刚添加成功的资源组,后面执行datax任务即使用刚添加成功的资源组进行调度运行。

任务配置

前提准备:查看hive表对应数据文件地址
TB1rJZmOVXXXXXXaVXXXXXXXXXX-645-410.png

操作步骤:

步骤1:前面ECS机器部署介绍的资源组配置好后,导航上点击进入“数据开发”页面进行任务开发。
TB18LwJOVXXXXXoaXXXXXXXXXXX-879-324.png

步骤2:创建工作流,命名为hive2MC_demo,选择一次性调度(本case是做一次性迁移,若需要每日生产调度则选择周期调度):
TB1lkIuOVXXXXbqapXXXXXXXXXX-1076-405.png

步骤3:在工作流中创建shell节点:
TB1G7cqOVXXXXa9aFXXXXXXXXXX-1084-515.png

步骤4:代码配置。Shell节点创建后,双击节点图,进入代码编辑器进行Shell代码编辑。

代码中进行datax配置,reader为hdfsreader,writer为odpswriter,具体代码如下,只需要修改reader和writer的配置:

shell_datax_home='/home/admin/shell_datax'
mkdir -p ${shell_datax_home}
shell_datax_config=${shell_datax_home}/${ALISA_TASK_ID}
echo '''
{
   "job": {
       "setting": {
           "speed": {
               "channel": 3
           }
       },
       "content": [
           {
               "reader": {
                   "name": "hdfsreader",
                   "parameter": {
                       "path": "/user/hive/warehouse/hive_table_name/*",
                       "defaultFS": "hdfs://xxx:port",
                       "column": [
                              {
                               "index": 0, 
                               "type": "date"
                              },
                              {
                               "index": 1,
                               "type": "STRING"
                              },
                              {
                               "index":2,
                               "type": "STRING"
                              },
                              {
                               "index":3,
                               "type": "STRING"
                              },
                              {
                               "index": 4,
                               "type": "long"
                              }
                              ,
                              {
                               "index": 5,
                               "type": "DOUBLE"
                              },
                              {
                               "index": 6,
                               "type": "long"
                              },
                              {
                               "index": 7,
                               "type": "long"
                              },
                              {
                               "index": 8,
                               "type": "long"
                              }
                       ],
                       "fileType": "text",
                       "encoding": "UTF-8",
                       "fieldDelimiter": ","
                   }

               },
               "writer": {
                   "name": "odpswriter",
                       "parameter": {
                         "project": "base_case", 
                         "table": "mc_dplus_good_sale", 
                         "partition":"pt=20160618", 
                         "column": ["create_time","good_cate","brand","buyer_id","trans_num","trans_amount","click_cnt","addcart_cnt","collect_cnt"],
                         "accessId": "xxxxxxxxxx",
                         "accessKey": "xxxxxxxxxxx",
                         "truncate": true,
                         "odpsServer": "http://service.odps.aliyun.com/api",
                         "tunnelServer": "http://dt.odps.aliyun.com",
                         "accountType": "aliyun"
                      }
               }
           }
       ]
   }
}''' > ${shell_datax_config}

echo "`date '+%Y-%m-%d %T'` shell datax config: ${shell_datax_config}"

/home/admin/datax3/bin/datax.py ${shell_datax_config}
shell_datax_run_result=$?

rm ${shell_datax_config}

if [ ${shell_datax_run_result} -ne 0 ]
then
   echo "`date '+%Y-%m-%d %T'` shell datax ended failed :("
   exit -1
fi
echo "`date '+%Y-%m-%d %T'` shell datax ended success~"

hdfsreader

  • path:hive表hdfs文件路径(注意格式);
  • defaultFS: hdfs文件系统namenode节点地址(注意格式);
  • index:hdfs数据文本第几列,以0开始。
  • Type:hdfs数据对应datax的类型,请看hdsf&datax类型对照表;
  • fileType: hdfs数据文件类型;
  • encoding: hdfs数据文件的编码;
  • fieldDelimiter: hdfs数据文件字段分隔符。
    Odpswriter
  • project: MaxCompute目标表所属项目名称;
  • table: MaxCompute目标表名称;
  • partition:表分区值,此case直接用固定值;
  • column:目标表列,注意顺序与reader的列顺序一一对应;
  • accessId:写入目标表所用的云账号access id;
  • accessKey:写入目标表所用的云长access key;
  • truncate:写入前是否清空当前表/分区数据,需要清空就true,需要保留就false。

步骤5:保存shell节点,提交工作流。
TB1JhEpOVXXXXasaFXXXXXXXXXX-998-323.png

步骤6:指派执行资源组。这里将给shell任务指定执行的机器即我们前面部署的ECS。
TB14HwROVXXXXcrXFXXXXXXXXXX-1356-508.png

hive2mc资源组即为《ECS机器部署》章节中配置的资源组。资源组配置好后,后面就可以执行shell任务进行数据迁移。

Step4:执行数据迁移任务

接前面的步骤,点击shell节点的操作->流程:

TB1JhEpOVXXXXasaFXXXXXXXXXX-998-323.png

TB1KnQuOVXXXXb_apXXXXXXXXXX-1265-310.png

进入该节点的任务管理视图界面,右击节点图,选择测试节点,即可把该阶段调度执行起来:

TB1Tj4tPXXXXXb5XXXXXXXXXXXX-1055-510.png

示例名称和业务日期采用默认,直接点击生成并运行,然后点击“前往查看运行结果”。

TB1m8szOVXXXXbtaXXXXXXXXXXX-585-286.png
进入测试实例视图,可以看到节点示例执行状态,如下图变成绿色打勾的符号说明执行成功,右击节点图选择 查看运行日志 查看具体执行日志。

TB1cy..OVXXXXatXXXXXXXXXXXX-810-339.png
查看日志中读出100条,写入失败0条,说明成功写入100条数据;

TB1WxQ4OVXXXXXnXpXXXXXXXXXX-1283-493.png

Step5:校验MaxCompute表数据

可以到数据开发中创建一个脚本文件

TB1fLoFOVXXXXaNaXXXXXXXXXXX-768-308.png
运行代码:

select create_time,good_cate,brand,buyer_id,trans_num,trans_amount,click_cnt,addcart_cnt,collect_cnt from mc_dplus_good_sale limit 10; 

查看数据是否正常。

TB1Zp31OVXXXXaMXpXXXXXXXXXX-1323-545.png
也可以执行

select count(*) from mc_dplus_good_sale where pt=20160618;

查看是否导入的数据是否是100条。

TB1IOUKOVXXXXX1XVXXXXXXXXXX-767-345.png

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
Maxcompute造数据-方法详解
造一点模拟数据的方法
3148 0
大数据近实时数据投递MaxCompute
本文介绍离线大数据场景使用MaxCompute构建云上近实时数仓,打通云下数据上云链路,解决数据复杂类型支持和动态分区问题,满足高级数据处理需求的最佳实践。
1414 0
大数据近实时数据投递MaxCompute
大数据近实时数据投递MaxCompute
4333 0
自建Hadoop迁移MaxCompute最佳实践
本文介绍如何将自建Hadoop集群及生态组件迁移到阿里云MaxCompute大数据服务
295 0
自建Hadoop迁移MaxCompute
客户基于ECS、IDC自建或在友商云平台自建了大数据集群,为了降低企业大数据计算平台的成本,提高大数据应用开发效率,更有效保障数据安全,把大数据集群的数据、作业、调度任务以及业务数据库整体迁移到MaxCompute和其他云产品。
741 0
TiDb和MaxCompute同步数据
一、环境准备 保证TiDb集群外网可以访问 DataWorks 二、开发 1、允许TiDb远程连接 登录到TiDb执行授权语句 grant all privileges on *.* to 'user'@'%' identified by 'password' with grant optio.
1522 0
大数据处理也要安全--关于MaxCompute的安全科普
企业从未像今天这样可以轻易地存储和使用大数据。然而,当您在使用大数据产品时,是否考虑过其中的安全问题呢?庆幸的是,阿里云产品专家和安全专家早就想你所想急你所急,先行一步将安全特性带入了大数据计算产品 MaxCompute。该产品在面世之初就具备强大的安全特性,让企业可以放心地在云上使用。
2383 0
理论与实践:如何从Hadoop迁移到MaxCompute
MaxCompute大数据计算服务,能提供快速、完全托管的PB级数据仓库解决方案,能够使用户经济且高效地分析处理海量数据。而用户往往之前使用了Hadoop实现大数据计算任务,在选择了阿里云大数据计算服务之后,如何从Hadoop向MaxCompute进行迁移就成为了一个需要面对的问题了。
3156 0
MaxCompute_2_MaxCompute数据迁移文档
乍一看标题会以为是不是作者写错了怎么会有从MaxCompute到MaxCompute迁移数据的场景呢在实际使用中已经有客户遇到了这种场景比如两个网络互通的专有云环境之间数据迁移、公共云数加DataIDE上两个云账号之间数据迁移、还有网络不通的两个MaxCompute项目数据迁移等等下面我们逐个场景介绍。
4534 0
+关注
隐林
阿里云大数据产品专家,擅长MaxCompute、机器学习、分布式、可视化、人工智能等大数据领域;
288
文章
38
问答
文章排行榜
最热
最新
相关电子书
更多
低代码开发师(初级)实战教程
立即下载
阿里巴巴DevOps 最佳实践手册
立即下载
冬季实战营第三期:MySQL数据库进阶实战
立即下载