本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 在 Aurora PostgreSQL-Compatible 上设置 Oracle UTL\$1FILE 功能 *Rakesh Raghav 和 anuradha chintha，Amazon Web Services* ## Summary 在从 Oracle 迁移到 Amazon Web Services (AWS) 云上的 Amazon Aurora PostgreSQL 兼容版的过程中，您可能会遇到多种挑战。例如，迁移依赖于 Oracle `UTL_FILE` 实用程序的代码始终是一项挑战。在 Oracle PL/SQL 中，`UTL_FILE` 软件包与底层操作系统一起用于文件操作，例如读取和写入。该 `UTL_FILE` 实用程序适用于服务器和客户机系统。  Amazon Aurora PostgreSQL-Compatible 的是一款托管式数据库产品。因此，无法访问 Database Server 上的文件。此模式将引导您完成整合 Amazon Simple Storage Service (Amazon S3) 和 Amazon Aurora PostgreSQL 兼容版，以实现 `UTL_FILE` 功能子网。使用此集成，我们可以创建和使用文件，而无需使用第三方提取、转换、加载（ETL）工具或服务。 或者，您可以设置亚马逊 CloudWatch 监控和亚马逊 SNS 通知。 建议在生产环境中实施该解决方案之前，对方案进行全面测试。 ## 先决条件和限制 **先决条件** + 一个有效的 Amazon Web Services account + AWS Database Migration Service (AWS DMS) 专业知识 + PL/pgSQL 编码方面的专业知识 + Amazon Aurora PostgreSQL-Compatible 集群 + 一个 S3 存储桶 **限制** 此模式不提供替代 Oracle `UTL_FILE` 实用程序的功能。但是，可以进一步增强步骤和示例代码，以实现数据库现代化目标。 **产品版本** + Amazon Aurora PostgreSQL-Compatible 版本 11.9。 ## 架构 **目标技术堆栈** + Amazon Aurora（兼容 PostgreSQL） + Amazon CloudWatch + Amazon Simple Notification Service（Amazon SNS） + Amazon S3 **目标架构** 下图高度概括此解决方案。 ![\[数据文件上传至 S3 存储桶，使用 aws_s3 扩展进行处理，然后发送到 Aurora 实例。\]](http://docs.aws.amazon.com/zh_cn/prescriptive-guidance/latest/patterns/images/pattern-img/3aeecd46-1f87-41f9-a9cd-f8181f92e83f/images/4a6c5f5c-58fb-4355-b243-d09a15c1cec6.png) 1. 文件将从应用程序上传至 S3 存储桶。 1. 该 `aws_s3` 扩展使用 PL/pgSQL 访问数据，并将数据上传到兼容 Aurora PostgreSQL 的版本。 ## 工具 + [Amazon Aurora PostgreSQL-Compatible](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/Aurora.AuroraPostgreSQL.html) – Amazon Aurora PostgreSQL-Compatible Edition 是一个完全托管式、兼容 PostgreSQL 和 ACID 的关系数据库引擎。您已了解了 MySQL 和 PostgreSQL 不仅具有高端商用数据库的速度和可靠性，同时还具有开源数据库的简单性和成本效益。 + [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-welcome.html) - AWS 命令行界面（AWS CLI）是用于管理 Amazon Web Services 的统一工具。只通过一个工具进行下载和配置，您就可以使用命令行控制多个 Amazon Web Services 并利用脚本来自动执行这些服务。 + [亚马逊 CloudWatch](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/WhatIsCloudWatch.html) — 亚马逊 CloudWatch 监控亚马逊 S3 的资源和使用情况。 + [Amazon S3](https://docs.aws.amazon.com/AmazonS3/latest/userguide/Welcome.html) – Amazon Simple Storage Service (Amazon S3) 是一项面向互联网的存储服务。Amazon S3 提供了一个存储层，用于接收和存储文件，以便在兼容 Aurora PostgreSQL 的集群之间使用和传输这些文件。 + [aws\$1s3](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/PostgreSQL.Procedural.Importing.html#aws_s3.table_import_from_s3) – 该 `aws_s3` 扩展程序集成了 Amazon S3 和 Aurora PostgreSQL-Compatible。 + [Amazon SNS](https://docs.aws.amazon.com/sns/latest/dg/welcome.html) – Amazon Simple Notification Service（Amazon SNS）可协调和管理发布者和客户端之间消息的传送或发送。在这种模式中，Amazon SNS 用于发送通知。 + [pgAdmin](https://www.pgadmin.org/docs/) – pgAdmin 是 Postgres 的开源管理工具。pgadmin 4 提供了一个用于创建、维护和使用数据库对象的图形界面。 **代码** 为了实现所需功能，该模式创建了多个与 `UTL_FILE` 命名类似的函数。*其他信息* 部分包含这些函数的代码库。 在代码中，将 `testaurorabucket` 替换为您的 S3 存储桶名称。将 `us-east-1` 替换为您测试 S3 存储桶所在的 AWS 区域。 ## 操作说明 ### 集成 Amazon S3 和 Aurora PostgreSQL-Compatible | Task | 说明 | 所需技能 | | --- | --- | --- | | 设置 IAM policy。 | 用于访问 S3 存储桶的 AWS Identity and Access Management (IAM) 角色。关于此代码，请参阅*更多信息*章节。 | AWS 管理员，数据库管理员 | | 将 Amazon S3 访问角色添加至 Aurora PostgreSQL。 | 创建两个 IAM 角色：一个角色用于对 Amazon S3 进行读取访问，一个角色用于对 Amazon S3 进行写入访问。将角色附加到兼容 Aurora PostgreSQL 的集群。 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_cn/prescriptive-guidance/latest/patterns/set-up-oracle-utl_file-functionality-on-aurora-postgresql-compatible.html)有关更多信息，请参阅 Aurora PostgreSQL-Compatible 文档，介绍如何将数据[导入](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_PostgreSQL.S3Import.html)和[导出](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/postgresql-s3-export.html)到 Amazon S3。 | AWS 管理员，数据库管理员 | ### 在 Aurora PostgreSQL-Compatible 中设置扩展 | Task | 说明 | 所需技能 | | --- | --- | --- | | 创建 aws\$1commons 扩展。 | `aws_commons` 扩展是 `aws_s3` 扩展的依赖项。 | 数据库管理员、开发人员 | | 创建 aws\$1s3 扩展。 | `aws_s3` 扩展程序与 Amazon S3 交互。 | 数据库管理员、开发人员 | ### 验证 Amazon S3 和 Aurora PostgreSQL-Compatible 的集成 | Task | 说明 | 所需技能 | | --- | --- | --- | | 将 Amazon S3 中的数据导入到 Aurora PostgreSQL | 要测试将文件导入到 Aurora PostgreSQL 兼容版中，请创建一个示例 CSV 文件并将其上传到 S3 存储桶中。基于 CSV 文件创建表定义，然后使用 `aws_s3.table_import_from_s3` 函数将文件加载到表中。 | 数据库管理员、开发人员 | | 测试将文件从 Aurora PostgreSQL 导出至 Amazon S3。 | 要测试从兼容 Aurora PostgreSQL 的导出文件，请创建一个测试表，在其中填充数据，然后使用 `aws_s3.query_export_to_s3` 函数导出数据。 | 数据库管理员、开发人员 | ### 若要模拟 UTL\$1FILE 实用程序，请创建包装函数 | Task | 说明 | 所需技能 | | --- | --- | --- | | 创建 utl\$1file\$1utility 架构。 | 该架构将包装器函数汇聚在一起。运行以下命令以创建 EIP。
CREATE SCHEMA utl_file_utility;
| 数据库管理员、开发人员 | | 创建 file\$1type 类型。 | 使用以下代码创建 `file_type` 类型。
CREATE TYPE utl_file_utility.file_type AS (
    p_path character varying(30),
    p_file_name character varying
);


| 数据库管理员/开发人员 | | 创建 init 函数 | `init` 函数初始化公共变量，例如 `bucket` 或 `region`。关于此代码，请参阅*更多信息*章节。 | 数据库管理员/开发人员 | | 创建包装器函数。 | 创建包装函数 `fopen`、`put_line` 和 `fclose`。关于代码，请参阅*更多信息*章节。 | 数据库管理员、开发人员 | ### 测试包装器函数 | Task | 说明 | 所需技能 | | --- | --- | --- | | 在写入模式测试包装器函数。 | 要在写入模式下测试包装器函数，请使用*其他信息*部分中提供的代码。 | 数据库管理员、开发人员 | | 在追加模式下测试包装器函数。 | 要在追加模式下测试包装器函数，请使用*其他信息*部分中提供的代码。 | 数据库管理员、开发人员 | ## 相关资源 + [Amazon S3 集成](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_PostgreSQL.S3Import.html) + [Amazon S3](https://aws.amazon.com/s3/) + [Aurora](https://aws.amazon.com/rds/aurora/?nc2=h_ql_prod_db_aa&aurora-whats-new.sort-by=item.additionalFields.postDateTime&aurora-whats-new.sort-order=desc) + [Amazon CloudWatch](https://aws.amazon.com/cloudwatch/) + [Amazon SNS](https://aws.amazon.com/sns/?nc2=h_ql_prod_ap_sns&whats-new-cards.sort-by=item.additionalFields.postDateTime&whats-new-cards.sort-order=desc) ## 附加信息 **设置 IAM policy** 创建以下策略。 | | | 策略名称 | JSON | | --- |--- | | S3 IntRead | 
{
    "Version": "2012-10-17",		 	 	 
    "Statement": [
        {
            "Sid": "S3integrationtest",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:ListBucket"
            ],
            "Resource": [
         "arn:aws:s3:::testaurorabucket/*",
         "arn:aws:s3:::testaurorabucket"
            ]
        }
    ]
}
| | S3 IntWrite | 
{
    "Version": "2012-10-17",		 	 	 
    "Statement": [
        {
            "Sid": "S3integrationtest",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",                
                "s3:ListBucket"
            ],
            "Resource": [                "arn:aws:s3:::testaurorabucket/*",                "arn:aws:s3:::testaurorabucket"
            ]
        }
    ]
}
| **创建 init 函数** 若要初始化常用变量，例如 `bucket` 或 `region`，请使用以下代码创建 `init` 函数。 ``` CREATE OR REPLACE FUNCTION utl_file_utility.init(     )     RETURNS void     LANGUAGE 'plpgsql'     COST 100     VOLATILE AS $BODY$ BEGIN       perform set_config       ( format( '%s.%s','UTL_FILE_UTILITY', 'region' )       , 'us-east-1'::text       , false );       perform set_config       ( format( '%s.%s','UTL_FILE_UTILITY', 's3bucket' )       , 'testaurorabucket'::text       , false ); END; $BODY$; ``` **创建包装器函数** 创建 `fopen`、`put_line` 和 `fclose` 包装器函数。 *fopen* ``` CREATE OR REPLACE FUNCTION utl_file_utility.fopen(     p_file_name character varying,     p_path character varying,     p_mode character DEFAULT 'W'::bpchar,     OUT p_file_type utl_file_utility.file_type)     RETURNS utl_file_utility.file_type     LANGUAGE 'plpgsql'     COST 100     VOLATILE AS $BODY$ declare     v_sql character varying;     v_cnt_stat integer;     v_cnt integer;     v_tabname character varying;     v_filewithpath character varying;     v_region character varying;     v_bucket character varying; BEGIN     /*initialize common variable */     PERFORM utl_file_utility.init();     v_region := current_setting( format( '%s.%s', 'UTL_FILE_UTILITY', 'region' ) );     v_bucket :=  current_setting( format( '%s.%s', 'UTL_FILE_UTILITY', 's3bucket' ) );         /* set tabname*/     v_tabname := substring(p_file_name,1,case when strpos(p_file_name,'.') = 0 then length(p_file_name) else strpos(p_file_name,'.') - 1 end );     v_filewithpath := case when NULLif(p_path,'') is null then p_file_name else concat_ws('/',p_path,p_file_name) end ;     raise notice 'v_bucket %, v_filewithpath % , v_region %', v_bucket,v_filewithpath, v_region;         /* APPEND MODE HANDLING; RETURN EXISTING FILE DETAILS IF PRESENT ELSE CREATE AN EMPTY FILE */     IF p_mode = 'A' THEN         v_sql := concat_ws('','create temp table if not exists ', v_tabname,' (col1 text)');         execute v_sql;         begin         PERFORM aws_s3.table_import_from_s3             ( v_tabname,             '',               'DELIMITER AS ''#''',             aws_commons.create_s3_uri             (     v_bucket,                 v_filewithpath ,                 v_region)             );         exception             when others then              raise notice 'File load issue ,%',sqlerrm;              raise;         end;         execute concat_ws('','select count(*) from ',v_tabname) into v_cnt;         IF v_cnt > 0         then             p_file_type.p_path := p_path;             p_file_type.p_file_name := p_file_name;         else                     PERFORM aws_s3.query_export_to_s3('select ''''',                             aws_commons.create_s3_uri(v_bucket, v_filewithpath, v_region)                                           );             p_file_type.p_path := p_path;             p_file_type.p_file_name := p_file_name;                 end if;         v_sql := concat_ws('','drop table ', v_tabname);                 execute v_sql;                 ELSEIF p_mode = 'W' THEN             PERFORM aws_s3.query_export_to_s3('select ''''',                             aws_commons.create_s3_uri(v_bucket, v_filewithpath, v_region)                                           );             p_file_type.p_path := p_path;             p_file_type.p_file_name := p_file_name;     END IF;         EXCEPTION         when others then             p_file_type.p_path := p_path;             p_file_type.p_file_name := p_file_name;             raise notice 'fopenerror,%',sqlerrm;             raise; END; $BODY$; ``` *put\$1line* ``` CREATE OR REPLACE FUNCTION utl_file_utility.put_line(     p_file_name character varying,     p_path character varying,     p_line text,     p_flag character DEFAULT 'W'::bpchar)     RETURNS boolean     LANGUAGE 'plpgsql'     COST 100     VOLATILE AS $BODY$ /************************************************************************** * Write line, p_line in windows format to file, p_fp - with carriage return * added before new line. **************************************************************************/ declare     v_sql varchar;     v_ins_sql varchar;     v_cnt INTEGER;     v_filewithpath character varying;     v_tabname  character varying;     v_bucket character varying;     v_region character varying;     BEGIN  PERFORM utl_file_utility.init(); /* check if temp table already exist */  v_tabname := substring(p_file_name,1,case when strpos(p_file_name,'.') = 0 then length(p_file_name) else strpos(p_file_name,'.') - 1 end );  v_sql := concat_ws('','select count(1) FROM pg_catalog.pg_class c LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace where n.nspname like ''pg_temp_%'''                          ,' AND pg_catalog.pg_table_is_visible(c.oid) AND Upper(relname) = Upper( '''                          ,  v_tabname ,''' ) ');    execute v_sql into v_cnt;     IF v_cnt = 0 THEN          v_sql := concat_ws('','create temp table ',v_tabname,' (col text)');         execute v_sql;         /* CHECK IF APPEND MODE */         IF upper(p_flag) = 'A' THEN             PERFORM utl_file_utility.init();                                     v_region := current_setting( format( '%s.%s', 'UTL_FILE_UTILITY', 'region' ) );             v_bucket :=  current_setting( format( '%s.%s', 'UTL_FILE_UTILITY', 's3bucket' ) );                         /* set tabname*/                         v_filewithpath := case when NULLif(p_path,'') is null then p_file_name else concat_ws('/',p_path,p_file_name) end ;                                     begin                PERFORM aws_s3.table_import_from_s3                      ( v_tabname,                           '',                          'DELIMITER AS ''#''',                         aws_commons.create_s3_uri                            ( v_bucket,                                v_filewithpath,                                v_region    )                     );             exception                 when others then                     raise notice  'Error Message : %',sqlerrm;                     raise;             end;             END IF;         END IF;     /* INSERT INTO TEMP TABLE */                   v_ins_sql := concat_ws('','insert into ',v_tabname,' values(''',p_line,''')');     execute v_ins_sql;     RETURN TRUE;     exception             when others then                 raise notice  'Error Message : %',sqlerrm;                 raise; END; $BODY$; ``` *fclose* ``` CREATE OR REPLACE FUNCTION utl_file_utility.fclose(     p_file_name character varying,     p_path character varying)     RETURNS boolean     LANGUAGE 'plpgsql'     COST 100     VOLATILE AS $BODY$ DECLARE     v_filewithpath character varying;     v_bucket character varying;     v_region character varying;     v_tabname character varying; v_sql character varying; BEGIN       PERFORM utl_file_utility.init();       v_region := current_setting( format( '%s.%s', 'UTL_FILE_UTILITY', 'region' ) );     v_bucket :=  current_setting( format( '%s.%s', 'UTL_FILE_UTILITY', 's3bucket' ) );     v_tabname := substring(p_file_name,1,case when strpos(p_file_name,'.') = 0 then length(p_file_name) else strpos(p_file_name,'.') - 1 end );     v_filewithpath := case when NULLif(p_path,'') is null then p_file_name else concat_ws('/',p_path,p_file_name) end ;     raise notice 'v_bucket %, v_filewithpath % , v_region %', v_bucket,v_filewithpath, v_region ;         /* exporting to s3 */     perform aws_s3.query_export_to_s3         (concat_ws('','select * from ',v_tabname,'  order by ctid asc'),             aws_commons.create_s3_uri(v_bucket, v_filewithpath, v_region)         );    v_sql := concat_ws('','drop table ', v_tabname);     execute v_sql;        RETURN TRUE; EXCEPTION        when others then      raise notice 'error fclose %',sqlerrm;      RAISE; END; $BODY$; ``` **测试您的设置与包装器功能** 使用下面的代码示例测试设置。 *测试写入模式* 以下代码在 S3 存储桶中写入名为 `s3inttest` 的文件。 ``` do $$ declare l_file_name varchar := 's3inttest' ; l_path varchar := 'integration_test' ; l_mode char(1) := 'W'; l_fs utl_file_utility.file_type ; l_status boolean; begin select * from utl_file_utility.fopen( l_file_name, l_path , l_mode ) into l_fs ; raise notice 'fopen : l_fs : %', l_fs; select * from utl_file_utility.put_line( l_file_name, l_path ,'this is test file:in s3bucket: for test purpose', l_mode ) into l_status ; raise notice 'put_line : l_status %', l_status; select * from utl_file_utility.fclose( l_file_name , l_path ) into l_status ; raise notice 'fclose : l_status %', l_status; end; $$ ``` *测试追加模式* 以下代码将行追加到上一个测试中创建的 `s3inttest` 文件上。 ``` do $$ declare l_file_name varchar := 's3inttest' ; l_path varchar := 'integration_test' ; l_mode char(1) := 'A'; l_fs utl_file_utility.file_type ; l_status boolean; begin select * from utl_file_utility.fopen( l_file_name, l_path , l_mode ) into l_fs ; raise notice 'fopen : l_fs : %', l_fs; select * from utl_file_utility.put_line( l_file_name, l_path ,'this is test file:in s3bucket: for test purpose : append 1', l_mode ) into l_status ; raise notice 'put_line : l_status %', l_status; select * from utl_file_utility.put_line( l_file_name, l_path ,'this is test file:in s3bucket : for test purpose : append 2', l_mode ) into l_status ; raise notice 'put_line : l_status %', l_status; select * from utl_file_utility.fclose( l_file_name , l_path ) into l_status ; raise notice 'fclose : l_status %', l_status; end; $$ ``` **Amazon SNS 通知** 或者，您可以在 S3 存储桶上设置亚马逊 CloudWatch 监控和 Amazon SNS 通知。有关更多信息，请参阅[监控 Amazon S3](https://docs.aws.amazon.com/AmazonS3/latest/userguide/monitoring-overview.html) 和[设置 Amazon SNS 通知](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/US_SetupSNS.html)。