使用OSS外表高速导入OSS数据 - 云原生数据仓库AnalyticDB PostgreSQL版

功能介绍

目前gpossext支持读写TEXT、CSV格式的文件以及GZIP压缩格式的TEXT、CSV文件。

gpossext架构图如下。

TEXT和CSV格式说明

下列几个参数可以在外表DDL参数中指定，用于规定读写OSS的文件格式：

TEXT和CSV行分割符号是\n，也就是换行符。
DELIMITER用于定义列的分割符：
- 当用户数据中包括DELIMITER时，则需要和QUOTE参数一同使用。
- 推荐的列分割符有,、\t、|或一些不常出现的字符。
QUOTE用于包裹有特殊字符的用户数据（以列为单位）：
- 包含有特殊字符的字符串会被QUOTE包裹，用于区分用户数据和控制字符。
- 如果不必要，例如整数，基于优化效率的考虑，不必使用QUOTE包裹数据。
- QUOTE不能和DELIMITER相同，默认QUOTE是双引号。
- 当用户数据中包含了QUOTE字符，则需要使用转义字符ESCAPE加以区分。
ESCAPE用于特殊字符转义：
- 转义字符出现在需要转义的特殊字符前，表示它不是一个特殊字符。
- ESCAPE默认和QUOTE相同，为双引号""。
- 也支持设置成\（MySQL默认的转义字符）或别的字符。

表 1. 典型的TEXT和CSV默认控制字符
控制字符和格式	TEXT	CSV
DELIMITER（列分割符）	\t（Tab）	, （Comma）
QUOTE（摘引）	" （Double-Quote）	"（Double-Quote）
ESCAPE（转义）	（不适用）	与QUOTE相同
NULL（空值）	\N（Backslash-N）	（无引号的空字符串）

说明所有的控制字符都必须是单字节字符。

注意事项

创建和使用外部表的语法，除了location相关参数，其余参数和Greenplum的使用方式相同。
数据导入导出的性能和AnalyticDB PostgreSQL的资源（CPU、I/O、内存、网络等）有关，也和OSS相关。为了获取最优的导入导出性能，建议在创建表时，使用列式存储加压缩功能。例如，指定子句“WITH (APPENDONLY=true, ORIENTATION=column, COMPRESSTYPE=zlib, COMPRESSLEVEL=5, BLOCKSIZE=1048576)，详细信息，请参见 Greenplum Database 表创建语法官方文档。
为了保证数据导入导出的性能，请保证OSS与AnalyticDB PostgreSQL在同一地域下。

操作步骤

创建OSS外部表插件。
使用OSS外部表时，需要在AnalyticDB PostgreSQL中先创建OSS外部表插件（每个库中均需要单独创建）。创建命令如下：
```
CREATE EXTENSION IF NOT EXISTS oss_ext;
```
将待导入AnalyticDB PostgreSQL的数据均匀分散存储在多个OSS文件中。操作方式，请参见大文件切分。

说明 AnalyticDB PostgreSQL的每个数据分区（Segment节点）将按轮询方式并行对OSS上的数据文件进行读取，文件的数目建议为数据节点数（Segment个数）的整数倍，从而提升读取效率。

在AnalyticDB PostgreSQL中，创建READABLE外部表。

创建OSS外部表语法如下。

CREATE [READABLE] EXTERNAL TABLE tablename
( columnname datatype [, ...] | LIKE othertable )
LOCATION ('ossprotocol')
FORMAT 'TEXT'
            [( [HEADER]
               [DELIMITER [AS] 'delimiter' | 'OFF']
               [NULL [AS] 'null string']
               [ESCAPE [AS] 'escape' | 'OFF']
               [NEWLINE [ AS ] 'LF' | 'CR' | 'CRLF']
               [FILL MISSING FIELDS] )]
           | 'CSV'
            [( [HEADER]
               [QUOTE [AS] 'quote']
               [DELIMITER [AS] 'delimiter']
               [NULL [AS] 'null string']
               [FORCE NOT NULL column [, ...]]
               [ESCAPE [AS] 'escape']
               [NEWLINE [ AS ] 'LF' | 'CR' | 'CRLF']
               [FILL MISSING FIELDS] )]
[ ENCODING 'encoding' ]
[ [LOG ERRORS [INTO error_table]] SEGMENT REJECT LIMIT count
       [ROWS | PERCENT] ]
ossprotocol:
   oss://oss_endpoint [prefix=prefix_name|dir=[folder/[folder/]...]/file_name|filepath=[folder/[folder/]...]/file_name]
    id=userossid key=userosskey bucket=ossbucket compressiontype=[none|gzip] async=[true|false]

参数说明如下。


参数	说明
FORMAT	支持文件格式，例如TEXT、CSV。
ENCODING	文件中数据的编码格式，例如UTF8。
LOG ERRORS	指定该子句可以忽略掉导入中出错的数据，将这些数据写入error_table，并可以使用count参数指定报错的阈值。说明通过`LOG ERRORS`将错误行信息记录到内部关联文件。 `LOG ERRORS SEGMENT REJECT LIMIT 5;` 通过函数`gp_read_error_log('external_table_name')`可以读取错误行信息。 `SELECT * FROM gp_read_error_log('external_table_name');` 内部文件随外表删除而删除，也可以通过函数`gp_truncate_error_log('external_table_name')`删除。 `SELECT gp_truncate_error_log('external_table_name');`
oss://oss_endpoint	协议和Endpoint，格式为`协议名://oss_endpoint`，其中协议名为oss，oss_endpoint为OSS对应区域的域名。示例如下： `oss://oss-cn-hangzhou.aliyuncs.com` 注意如果是从阿里云的主机访问数据库，应该使用内网域名（即带有`internal`的域名），避免产生公网流量。
id	阿里云账号的AccessKey ID。获取AccessKey操作，请参见获取AccessKey。
key	阿里云账号的AccessKey Secret。获取AccessKey操作，请参见获取AccessKey。
bucket	指定数据文件所在的Bucket，需要在OSS上预先创建。
prefix	指定数据文件对应路径名的前缀，不支持正则表达式，仅是匹配前缀。说明与dir、filepath互斥，三者只能设置其中一个。 READABLE外部表会在导入数据时将含有这一前缀的所有OSS文件都会被导入。如果指定prefix=test/filename，以下文件都会被导入： test/filename test/filenamexxx test/filename/aa test/filenameyyy/aa test/filenameyyy/bb/aa 如果指定prefix=test/filename/，只有以下文件会被导入（上面列的其他文件不会被导入）： test/filename/aa
dir	OSS中的虚拟文件夹路径。说明与prefix、filepath互斥，三者只能设置其中一个。文件夹路径需要以`/`结尾，如`test/mydir/`。在导入数据时，使用此参数创建外部表，会导入指定虚拟目录下的所有文件，但不包括它子目录和子目录下的文件。与filepath不同，dir下的文件没有命名要求。
filepath	OSS中包含路径的文件名称。说明与prefix、dir互斥，三者只能设置其中一个。这个参数只能在创建READABLE外部表时指定，即仅支持在导入数据时使用。
compressiontype	导入文件的压缩格式。 none（默认值）：导入的文件未压缩。 gzip：导入的文件压缩格式为GZIP。说明目前仅支持GZIP压缩格式。
compressionlevel	设置写入OSS的文件的压缩等级，取值范围为1~9，默认值为6。示例如下： `compressionlevel=6`
oss_connect_timeout	设置连接超时。单位为秒，默认为10秒。
oss_dns_cache_timeout	设置DNS超时。单位为秒，默认为60秒。
oss_speed_limit	设置触发超时的最小速率。默认为1024字节，即1 KB。需要与oss_speed_time参数配合使用。说明如果使用默认值且连续15秒的传输速率小于1 KB，会触发超时。具体信息，请参见OSS SDK 错误处理。
oss_speed_time	设置触发超时的最长时间。默认为15秒。需要与oss_speed_limit参数配合使用。说明如果使用默认值且连续15秒的传输速率小于1 KB，会触发超时。具体信息，请参见OSS SDK 错误处理。
async	是否启用异步模式导入数据。开启辅助线程从OSS导入数据，加速导入性能。默认情况下异步模式为开启状态，如果需要关掉，可以使用参数`async = false`或`async = f`。异步模式和普通模式比，会消耗更多的硬件资源。

并行导入数据。
在AnalyticDB PostgreSQL数据库中执行如下命令，将OSS上的数据并行导入到AnalyticDB PostgreSQL数据库。
```
INSERT INTO <目标表> SELECT * FROM <外部表>
```

操作示例

本文以目标表example为例，介绍将OSS的数据通过外部表导入目标表example。

创建OSS外部表插件。
创建命令如下：
```
CREATE EXTENSION IF NOT EXISTS oss_ext;
```

创建目标表，用于装载数据。

CREATE TABLE example
        (date text, time text, open float,
         high float, low float, volume int)
         DISTRIBUTED BY (date);

创建OSS导入外部表。

创建外部表时，使用prefix参数指定待导入数据所在的路径。示例如下：

CREATE READABLE EXTERNAL TABLE ossexample
        (date text, time text, open float, high float,
        low float, volume int)
        location('oss://oss-cn-hangzhou.aliyuncs.com
        prefix=osstest/example id=XXX
        key=XXX bucket=testbucket compressiontype=gzip')
        FORMAT 'csv' (QUOTE '''' DELIMITER E'\t')
        ENCODING 'utf8'
        LOG ERRORS SEGMENT REJECT LIMIT 5;

创建外部表时，使用dir参数指定待导入数据所在的路径。示例如下：

CREATE READABLE EXTERNAL TABLE ossexample
        (date text, time text, open float, high float,
        low float, volume int)
        location('oss://oss-cn-hangzhou.aliyuncs.com
        dir=osstest/ id=XXX
        key=XXX bucket=testbucket')
        FORMAT 'csv'
        LOG ERRORS SEGMENT REJECT LIMIT 5;

创建外部表时，使用filepath参数指定待导入数据所在的路径。示例如下：

CREATE READABLE EXTERNAL TABLE ossexample
        (date text, time text, open float, high float,
        low float, volume int)
        location('oss://oss-cn-hangzhou.aliyuncs.com
        filepath=osstest/example.csv id=XXX
        key=XXX bucket=testbucket')
        FORMAT 'csv'
        LOG ERRORS SEGMENT REJECT LIMIT 5;

将数据并行从ossexample外部表导入example表中。
```
INSERT INTO example SELECT * FROM ossexample;
```

执行如下查询计划，可以看到每个Segment节点都会从OSS并行拉取数据，然后通过执行节点Redistribution Motion将数据HASH计算后分发给对应的Segment节点，接收数据的Segment节点通过INSERT执行节点进行入库。

EXPLAIN INSERT INTO example SELECT * FROM ossexample;
                                            QUERY PLAN
-----------------------------------------------------------------------------------------------
 Insert (slice0; segments: 4)  (rows=250000 width=92)
   ->  Redistribute Motion 4:4  (slice1; segments: 4)  (cost=0.00..11000.00 rows=250000 width=92)
         Hash Key: ossexample.date
         ->  External Scan on ossexample  (cost=0.00..11000.00 rows=250000 width=92)
(4 rows)

SDK错误处理

当导入或导出操作出错时，错误日志可能会出现如下信息：

code：出错请求的HTTP状态码。
error_code：OSS的错误码。
error_msg：OSS的错误信息。
req_id：标识该次请求的UUID。当您无法解决问题时，可以凭req_id来请求OSS开发工程师的帮助。

具体信息，请参见OSS API 错误响应，超时相关的错误可以使用oss_ext相关参数处理。

云原生数据仓库AnalyticDB PostgreSQL版：使用OSS外表高速导入OSS数据