首先，创建我们要用到的插件。

  2. CREATE OR REPLACE FUNCTION rds_oss_fdw_load_data_prepare(t_from text, t_to text)
  3.   RETURNS bool AS
  4. $BODY$
  5. DECLARE
  6.     t_exist  int;
  7.     curs1 refcursor;
  8.     r    record;
  9.     filepath text;
  10.     fileindex int8;
  11.     s1 text;
  12.     s2 text;
  13.     s3 text;
  14.     c int = 0;
  15.     s4 text;
  16.     s5 text;
  17.     ss4 text;
  18.     ss5 text;
  19.     sql text;
  20. BEGIN
  21.     create table if not exists oss_fdw_load_status(id BIGSERIAL primary key, filename text, size int8, rows int8 default 0, status int default 0);
  23.     select count(*) into t_exist from oss_fdw_load_status;
  25.     if t_exist != 0 then
  26.         RAISE NOTICE 'oss_fdw_load_status not empty';
  27.         return false;
  28.     end if;
  30.     -- 通过 oss_fdw_list_file 函数，把外部表 t_from 匹配的 OSS 中的文件列到表中
  31.     insert into oss_fdw_load_status (filename, size) select name,size from oss_fdw_list_file(t_from) order by size desc;
  33.     select count(*) into t_exist from oss_fdw_load_status;
  34.     if t_exist = 0 then
  35.         RAISE NOTICE 'oss_fdw_load_status empty,not task found';
  36.         return false;
  37.     end if;
  39.     return true;
  40. END;
  41. $BODY$
  42.     LANGUAGE plpgsql;
  44. -- 数据装载的工作函数
  45. CREATE OR REPLACE FUNCTION rds_oss_fdw_load_data_execute(t_from text, t_to text, num_work int, pass text)
  46.   RETURNS bool AS
  47. $BODY$
  48. DECLARE
  49.     t_exist  int;
  50.     curs1 refcursor;
  51.     r    record;
  52.     filepath text;
  53.     fileindex int8;
  54.     s1 text;
  55.     s2 text;
  56.     s3 text;
  57.     s4 text;
  58.     s5 text;
  59.     ss4 text;
  60.     ss5 text;
  61.     sql text;
  62.     db text;
  64. BEGIN
  65.     select count(*) into t_exist from oss_fdw_load_status;
  66.     if t_exist = 0 then
  67.         RAISE NOTICE 'oss_fdw_load_status empty';
  68.         return false;
  69.     end if;
  71.     s4 = 'oss_loader';
  72.     s5 = 'idle';
  73.     ss4 = '''' || s4 ||'''';
  74.     ss5 = '''' || s5 ||'''';
  75.     sql = 'select count(*) from pg_stat_activity where application_name = ' || ss4 || ' and state != ' || ss5;
  77.     select current_database() into db;
  78.     select current_user into user;
  80.     -- 通过游标，不断获取单个任务
  81.     OPEN curs1 FOR SELECT id, filename FROM oss_fdw_load_status order by id;
  82.     loop
  83.         fetch curs1 into r;
  84.         if not found then
  85.             exit;
  86.         end if;
  87.         fileindex = r.id;
  88.         filepath = r.filename;
  90.         s1 = '''' || t_from ||'''';
  91.         s2 = '''' || t_to ||'''';
  92.         s3 = '''' || filepath ||'''';
  94.         LOOP
  95.             -- 查看当前正在工作的任务数，过达到并发数就在这里等待
  96.             select a into c from dblink('dbname='||db ||' user='||user || ' password='||pass ,sql)as t(a int);
  97.             IF c < num_work THEN
  98.                 EXIT;
  99.             END IF;
  100.             RAISE NOTICE 'current runing % loader', c;
  101.             perform pg_sleep(1);
  102.         END LOOP;
  104.         -- 通过 DBLINK 创建异步任务
  105.         perform dis_conn('oss_loader_'||fileindex);
  106.         perform dblink_connect('oss_loader_'||fileindex, 'dbname='||db ||' user='||user || ' application_name=oss_loader' || ' password='||pass);
  107.         perform dblink_send_query('oss_loader_'||fileindex, format('
  108.             begin;
  109.             select rds_oss_fdw_load_single_file(%s,%s,%s,%s);
  110.             end;'
  111.             , fileindex, s1, s2, s3)
  112.         );
  113.         RAISE NOTICE 'runing loader task % filename %',fileindex, filepath;
  114.     end loop;
  115.     close curs1;
  117.     -- 任务分配完成，等待所有任务完成
  118.     LOOP
  119.         select a into c from dblink('dbname='||db ||' user='||user || ' password='||pass ,sql)as t(a int);
  120.         IF c = 0 THEN
  121.             EXIT;
  122.         END IF;
  123.         perform pg_sleep(1);
  124.     END LOOP;
  126.     return true;
  127. END;
  128. $BODY$
  129.     LANGUAGE plpgsql;
  131. -- 单个文件的数据装在函数
  132. CREATE OR REPLACE FUNCTION rds_oss_fdw_load_single_file(taskid int8, t_from text, t_to text, filepath text)
  133.   RETURNS void AS
  134. $BODY$
  135. DECLARE
  136.     rowscount int8 = 0;
  137.     current text;
  138.     sql text;
  139. BEGIN
  140.     -- 配置 GUC 参数，指定要导入的 OSS 上的文件
  141.     perform set_config('oss_fdw.rds_read_one_file',filepath,true);
  142.     select current_setting('oss_fdw.rds_read_one_file') into current;
  143.     RAISE NOTICE 'begin load %', current;
  145.     -- 通过动态 SQL 导入数据
  146.     EXECUTE 'insert into '|| t_to || ' select * from ' || t_from;
  147.     GET DIAGNOSTICS rowscount = ROW_COUNT;
  149.     -- 导入完成后，把结果保存到状态表中
  150.     RAISE NOTICE 'end load id % % to % % rows', taskid, filepath, t_to, rowscount;
  151.     update oss_fdw_load_status set rows = rowscount,status = 1 where id = taskid;
  152.     return;
  154. EXCEPTION
  155.     when others then
  156.     RAISE 'run rds_oss_fdw_load_single_file with error';
  157. END;
  158. $BODY$
  159.     LANGUAGE plpgsql;
  161. -- 关闭连接不报错
  162. create or replace function dis_conn(name) returns void as $$  
  163. declare  
  164. begin  
  165.   perform dblink_disconnect($1);  
  166.   return;  
  167. exception when others then  
  168.   return;  
  169. end;  
  170. $$ language plpgsql strict;

执行后，会看到表 oss_fdw_load_status 中，保存了准备导入的所有文件列表，用户可以做适当的删减定制。

2. 数据装载

  1.  select rds_oss_fdw_load_data_execute('oss_table','lineitem',10,'mypassword');

期间，我们可以通过下列 SQL 查看正在工作的异步会话状态

4.管理状态

同时，我们也可以随时中断数据导入工作

  1. select pg_terminate_backend(pid),application_name, state ,query from pg_stat_activity where state != 'idle' and pid != pg_backend_pid() and application_name='oss_loader';

6. 性能

使用 TPCC 100GB的数据进行装载测试，耗时 10 分钟，平均 170MB/S

  1. select rds_oss_fdw_load_data_prepare('t_oss2','lineitem');
  3. select rds_oss_fdw_load_data_execute('t_oss2','lineitem',10,'123456Zwj');
  5. select sum(size)/1024/1024 from oss_fdw_load_status;
  6.       ?column?      
  7. --------------------
  8.  22561.919849395752
  9. (1 row)
  11. select pg_size_pretty(pg_relation_size(oid)) from pg_class where relname = 'lineitem';
  12.  pg_size_pretty 
  13. ----------------
  14.  101 GB
  15. (1 row)

本文使用 plsql + dblink 的方式加速了 OSS 的数据导入。另外，大家也可以关注到以下三点

  • 1. PostgreSQL 默认的过程语言 pl/pgsql 相当好用，和 SQL 引擎紧密结合且学习成本低。我们推荐用户把业务逻辑用它实现。使用过程语言相对于在客户端执行 SQL，消除了服务器到和客户端的网络开销，有天然的性能优势。
  • 3. 阿里云开发的 oss_fdw 能在 PostgreSQL 和 OSS 之间做快速的数据交换。oss_fdw 支持 CSV 和压缩方式 CSV 数据的读和写，且很容易用并行加速。oss_fdw 的性能相对于 jdbc insert 和 copy 有压倒的性能优势。