本文向您介绍如何通过DataWorks中的数据集成创建数据同步任务,对MaxCompute中的数据进行导入和导出操作。
前置条件
添加MaxCompute数据源
若需要添加的数据源是当前MaxCompute项目,则无需进行此操作。该项目创建成功时,即在数据集成的数据源中,默认将该项目添加为数据源,数据源名为odps_first。
- 以项目管理员身份进入DataWorks管理控制台,单击项目列表下对应项目操作栏中的进入工作区。
说明 只有项目管理员角色才能够新建数据源,其他角色的成员仅能查看数据源。
- 单击顶部菜单栏中的数据集成。
- 单击左侧导航栏的数据源,进入数据源页面。
- 单击新增数据源,选择数据源类型为MaxCompute(ODPS)。
- 在新增数据源弹出框中填写相关配置项。
配置项说明如下:
- 数据源名称: 由英文字母、数字、下划线组成且需以字符或下划线开头,长度不超过60个字符。
- 数据源描述: 对数据源进行简单描述,不得超过80个字符。
- ODPS Endpoint: 默认只读。从系统配置中自动读取。
- Tunnel Endpoint:MaxCompute Tunnel服务的连接地址,详情请参见配置Endpoint。
- ODPS项目名称: 对应的MaxCompute Project标识。
- AccessKey ID: 与MaxCompute Project Owner云账号对应的AccessKeyID。
- AccessKey Secret: 与MaxCompute Project Owner云账号对应的AccessKeySecret,与AccessKeyID成对使用。
- 单击测试连通性。
- 若测试连通性成功,单击完成即可。
说明 其他的数据源的配置请参见支持的数据源与读写插件。
通过数据集成导入数据
以将MySQL的数据导入到MaxCompute中为例,您可以通过向导模式和脚本模式两种方式配置同步任务。
向导模式配置同步任务
- 新建向导模式的同步任务。
- 选择来源。
选择MySQL数据源及源头表mytest,数据浏览默认是收起的,选择后单击下一步。
- 选择目标。
目标即MaxCompute(原ODPS),表可以是提前创建,也可以在此处单击快速建表。
配置项说明:- 分区信息:必须指定到最后一级分区。例如把数据写入一个三级分区表,必须配置到最后一级分区,例如pt=20150101,type=1,biz=2。非分区表无此项配置。
- 清理规则:
- 写入前清理已有数据:导数据之前,清空表或者分区的所有数据,相当于insert overwrite。
- 写入前保留已有数据:导数据之前不清理任何数据,每次运行数据都是追加进去的,相当于insert into。
- 映射字段。
选择字段的映射关系。需对字段映射关系进行配置,左侧源头表字段和右侧目标表字段为一一对应的关系。
- 通道控制。
单击下一步,配置作业速率上限和脏数据检查规则。
配置项说明如下:- 作业速率上限:是指数据同步作业可能达到的最高速率,其最终实际速率受网络环境、数据库配置等的影响。
- 作业并发数:从单同步作业来看,作业并发数*单并发的传输速率=作业传输总速率。
当作业速率上限已选定的情况下,应该如何选择作业并发数?- 如果您的数据源是线上的业务库,建议您不要将并发数设置过大,以防对线上库造成影响。
- 如果您对数据同步速率特别在意,建议您选择最大作业速率上限和较大的作业并发数。
- 预览保存。
完成上述配置后,上下滚动鼠标可查看任务配置,如若无误,单击保存。
运行同步任务
直接运行同步任务
只有在同步任务设置了系统变量参数,在运行时才会自动弹出弹框配置参数变量。
同步任务保存后,直接单击运行,任务便会立刻运行。您也可以单击提交,将同步任务提交到DataWorks的调度系统中,调度系统会按照配置属性在从第二天开始自动定时执行。
脚本模式配置同步任务
您可参考以下脚本进行配置同步任务操作,其他配置与任务运行同向导模式方式。
{
"type": "job",
"version": "1.0",
"configuration": {
"reader": {
"plugin": "mysql",
"parameter": {
"datasource": "mysql",
"where": "",
"splitPk": "id",
"connection": [
{
"table": [
"person"
],
"datasource": "mysql"
}
],
"connectionTable": "person",
"column": [
"id",
"name"
]
}
},
"writer": {
"plugin": "odps",
"parameter": {
"datasource": "odps_first",
"table": "a1",
"truncate": true,
"partition": "pt=${bdp.system.bizdate}",
"column": [
"id",
"col1"
]
}
},
"setting": {
"speed": {
"mbps": "1",
"concurrent": "1"
}
}
}
}