350      人物

5.1 数据的导入__第五章 Data Pipeline_使用手册_分析型数据库-阿里云

分析型数据库支持多种数据入库方式，包括但不限于：

（1）内置将MaxCompute中的海量数据快速批量导入；

（2）支持标准的insert/delete语法，可使用用户程序、Kettle等第三方工具写入实时写入表；

（3）支持阿里云数据集成(CDP)，将各类数据源导入批量导入表或实时写入表。阿里云大数据开发平台；

（4）支持阿里云数据传输(DTS)，从阿里云RDS实时同步数据变更到分析型数据库。

本章节主要描述（1）和（3）两种数据导入方式，其余导入方式详见本手册第八章。

SQL方式将MaxCompute数据导入

分析型数据库目前内置支持从阿里云外售的MaxCompute（原ODPS）中导入数据。

将MaxCompute数据导入分析型数据库有几种方法：直接使用SQL命令进行导入，或通过MDS for AnalyticDB界面进行导入、通过数据集成（原CDP）配置job进行导入。

通过DMS for AnalyticDB界面导入数据已在《快速入门》进行了介绍；通过数据集成导入数据可以参考下一小章节。下面主要详细介绍通过SQL命令将MaxCompute数据导入AnalyticDB。

步骤一： 准备MaxCompute表或分区。

首先我们需要创建好数据源头的MaxCompute表（可以是分区表或非分区表，目前要求AnalyticDB中的字段名称和MaxCompute表中的字段名称一致），并在表中准备好要导入的数据。

如在MaxCompute中创建一个表：

use projecta；--在MaxCompute的某个project中创建表
CREATE TABLE 
odps2ads_test (
user_id bigint , 
amt bigint , 
num bigint , 
cat_id bigint , 
thedate bigint 
)
PARTITIONED BY(dt STRING);

往表中导入数据并创建分区，如：dt=’20160808’。

步骤二：账号授权。

首次导入一个新的MaxCompute表时，需要在MaxCompute中将表Describe和Select权限授权给AnalyticDB的导入账号。公共云导入账号为garuda_build@aliyun.com以及garuda_data@aliyun.com（两个都需要授权）。各个专有云的导入账号名参照专有云的相关配置文档，一般为test1000000009@aliyun.com。

授权命令：

USE projecta；--表所属ODPS project
ADD USER ALIYUN$xxxx@aliyun.com;--输入正确的云账号
GRANT Describe,Select ON TABLE table_name TO USER ALIYUN$xxxx@aliyun.com;--输入需要赋权的表和正确的云账号

另外为了保护用户的数据安全，AnalyticDB目前仅允许导入操作者为Proejct Owner的ODPS Project的数据，或者操作者为MaxCompute表的owner（大部分专有云无此限制）。

步骤三：准备AnalyticDB表，注意创建时更新方式属性为“批量更新”，同时把表的Load Data权限授权给导入操作者（一把表创建者都默认有该权限）。

如表：

CREATE TABLE db_name.odps2ads_test  ( 
user_id bigint,
amt int,
num int,
cat_id int,
thedate int,
primary key (user_id)
)
PARTITION BY HASH KEY(user_id) 
PARTITION NUM 40
TABLEGROUP group_name
options (updateType='batch');
--注意指定好数据库名和表组名

步骤四:在AnalyticDB中通过SQL命令导入数据。

语法格式：

LOAD DATA 
FROM 'sourcepath'
[IN VERSION dataVersion]
[OVERWRITE] INTO TABLE tablename [PARTITION (partition_name,...)]

说明：

• 如果使用MaxCompute（原ODPS）数据源，则sourcepath为：

    odps://<project>/<table>/[<partition-1=v1>/.../<partition-n=vn>]

  • ODPS项目名称 project 。
  • ODPS表名 table 。
  • ODPS分区 partition-n=vn ，可以是多级任意类型分区。

• 数据版本dataVersion，是分析型数据库管理一次导入数据用的批次号，可用于未来可能开放的数据回滚等feature，通常情况下可以不需要填写。

  • 数据版本必须是long型，且取值在表上单调递增。
  • 如果不指定则取当前时间（秒），格式是 yyyyMMddHHmss ，如 20140812080000。
• 覆盖导入选项（ OVERWRITE ），导入时如果指定数据日期的表在分析型数据库线上已存在，则返回异常，除非显示指定覆盖。
• 分析型数据库表名（ tablename ），格式 table_schema.table_name ，其中 table_name 是分析型数据库表名， table_schema 是表所属DB名，表名不区分大小写。

• PARTITION，分析型数据库表分区，分区格式 partition_column=partition_value ，其中 partition_column 是分区列名， partition_value 是分区值

  • 分区值必须是 long 型；
  • 分区值不存在时表示动态分区，例如第一级hash分区；
  • 不区分大小写；
  • 目前最多支持二级分区。

• 执行返回值： ‘‘、,任务ID，用于后续查询导入状态.任务ID是唯一标识该导入任务，返回的是字符串，最长256字节。

示例：

LOAD DATA 
FROM 'odps://<project>/odps2ads_test/dt=20160808'
INTO TABLE db_name.odps2ads_test
--注意<project> 填写MaxCompute表所属的项目名称
`

步骤五：查看成功导入的数据。

通过数据集成(CDP)将RDS等其他数据源的数据导入

当表的数据源是RDS、OSS等其它的云系统，我们可以通过阿里云的数据集成（原CDP）产品进行数据同步。

批量更新表导入

专有云上批量导入

专有云上可以通过大数据开发套件的数据同步任务进行操作，数据同步任务即通过封装数据集成实现数据导入。具体步骤请看“大数据开发套件”的用户指南相关数据源配置和数据同步任务配置章节。

前提条件：

• 分析型数据库目标表更新方式是“批量更新”。
• 在分析型数据库中给MaxCompute的base_meta这个project的owner账号至少授予表的Load Data权限，base_meta的owner账号信息可以在CMDB中查到。

注意“大数据开发套件”中数据同步任务目标为ads数据源的“导入模式”配置项需要选择“批量导入”。

公共云上批量导入

公共云上可在 https://www.aliyun.com/product/cdp/ 上开通数据集成（可能需要申请公测），

前提条件：

• 分析型数据库目标表更新方式是“批量更新”。
• 在分析型数据库中给cloud-data-pipeline@aliyun-inner.com这个账号至少授予表的Load Data权限。

数据集成配置示例：

{
    "type": "job",
    "traceId": "rds to ads job test",
    "version": "1.0",
    "configuration": {
        "setting": {
        },
        "reader": {
            "plugin": "mysql",
            "parameter": {
                "instanceName": "你的RDS的实例名",
                "database": "RDS数据库名",
                "table": "RDS表名",
                "splitPk": "任意一个列的名字",
                "username": "RDS用户名",
                "password": "RDS密码",
                "column": ["*"],
            }
        },
        "writer": {
          "plugin": "ads",
          "parameter": {
            "url": "在分析型数据库的控制台中选择数据库时提供的连接信息",
            "schema": "分析型数据库数据库名",
            "table": "分析型数据库表名",
            "username": "你的access key id",
            "password": "你的access key secret",
            "partition": "",
            "lifeCycle": 2,
            "overWrite": true
            }
        }
    }
}

运行该job即可。

结束批量导入任务

用户可以自行结束批量导入任务，运行如下命令：

kill loaddata 'jobId'

"writer": {
          "name": "adswriter",
          "parameter": {
            "writeMode": "insert",
            "url": "在分析型数据库的控制台中选择数据库时提供的连接信息",
            "schema": "分析型数据库数据库名",
            "table": "分析型数据库表名",
            "column": ["*"],
            "username": "你的access key id",
            "password": "你的access key secret",
            "partition": "id,ds=2015"
          }