数据库内核月报－ 2018/02 - PgSQL · 最佳实践 · 利用异步 dblink 快速从 oss 装载数据 - 《数据库内核月报》

摘要
一.准备工作
二.创建异步化存储过程
三.使用函数装载数据
总结

摘要

总所周知，阿里云的 PostgreSQL 和 HybridDB for PostgreSQL 和 oss 是全面互通的。 HybridDB for PostgreSQL 由于是 MPP 架构天生包括多个计算节点，能够以为并发的方式读写 oss 上的数据。PostgreSQL 在这方面要差一点，默认情况下只能单进程读写 OSS，不过通过 dblink 的加持，我们也能让 OSS 中的数据快速装载到 PostgreSQL。本文就给大家讲讲这其中的黑科技。

一.准备工作

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


1. create extension dblink;
2. create extension oss_fdw;

二.创建异步化存储过程


1. -- 异步数据装载的准备工作
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. c int = 0;
58. s4 text;
59. s5 text;
60. ss4 text;
61. ss5 text;
62. sql text;
63. db text;
64. user text;
65. BEGIN
66. select count(*) into t_exist from oss_fdw_load_status;
67. if t_exist = 0 then
68. RAISE NOTICE 'oss_fdw_load_status empty';
69. return false;
70. end if;

72. s4 = 'oss_loader';
73. s5 = 'idle';
74. ss4 = '''' || s4 ||'''';
75. ss5 = '''' || s5 ||'''';
76. sql = 'select count(*) from pg_stat_activity where application_name = ' || ss4 || ' and state != ' || ss5;

78. select current_database() into db;
79. select current_user into user;

81. -- 通过游标，不断获取单个任务
82. OPEN curs1 FOR SELECT id, filename FROM oss_fdw_load_status order by id;
83. loop
84. fetch curs1 into r;
85. if not found then
86. exit;
87. end if;
88. fileindex = r.id;
89. filepath = r.filename;

91. s1 = '''' || t_from ||'''';
92. s2 = '''' || t_to ||'''';
93. s3 = '''' || filepath ||'''';

95. LOOP
96. -- 查看当前正在工作的任务数，过达到并发数就在这里等待
97. select a into c from dblink('dbname='||db ||' user='||user || ' password='||pass ,sql)as t(a int);
98. IF c < num_work THEN
99. EXIT;
100. END IF;
101. RAISE NOTICE 'current runing % loader', c;
102. perform pg_sleep(1);
103. END LOOP;

105. -- 通过 DBLINK 创建异步任务
106. perform dis_conn('oss_loader_'||fileindex);
107. perform dblink_connect('oss_loader_'||fileindex, 'dbname='||db ||' user='||user || ' application_name=oss_loader' || ' password='||pass);
108. perform dblink_send_query('oss_loader_'||fileindex, format('
109. begin;
110. select rds_oss_fdw_load_single_file(%s,%s,%s,%s);
111. end;'
112. , fileindex, s1, s2, s3)
113. );
114. RAISE NOTICE 'runing loader task % filename %',fileindex, filepath;
115. end loop;
116. close curs1;

118. -- 任务分配完成，等待所有任务完成
119. LOOP
120. select a into c from dblink('dbname='||db ||' user='||user || ' password='||pass ,sql)as t(a int);
121. IF c = 0 THEN
122. EXIT;
123. END IF;
124. RAISE NOTICE 'current runing % loader', c;
125. perform pg_sleep(1);
126. END LOOP;

128. return true;
129. END;
130. $BODY$
131. LANGUAGE plpgsql;

133. -- 单个文件的数据装在函数
134. CREATE OR REPLACE FUNCTION rds_oss_fdw_load_single_file(taskid int8, t_from text, t_to text, filepath text)
135. RETURNS void AS
136. $BODY$
137. DECLARE
138. rowscount int8 = 0;
139. current text;
140. sql text;
141. BEGIN
142. -- 配置 GUC 参数，指定要导入的 OSS 上的文件
143. perform set_config('oss_fdw.rds_read_one_file',filepath,true);
144. select current_setting('oss_fdw.rds_read_one_file') into current;
145. RAISE NOTICE 'begin load %', current;

147. -- 通过动态 SQL 导入数据
148. EXECUTE 'insert into '|| t_to || ' select * from ' || t_from;
149. GET DIAGNOSTICS rowscount = ROW_COUNT;

151. -- 导入完成后，把结果保存到状态表中
152. RAISE NOTICE 'end load id % % to % % rows', taskid, filepath, t_to, rowscount;
153. update oss_fdw_load_status set rows = rowscount,status = 1 where id = taskid;
154. return;

156. EXCEPTION
157. when others then
158. RAISE 'run rds_oss_fdw_load_single_file with error';
159. END;
160. $BODY$
161. LANGUAGE plpgsql;

163. -- 关闭连接不报错
164. create or replace function dis_conn(name) returns void as $$
165. declare
166. begin
167. perform dblink_disconnect($1);
168. return;
169. exception when others then
170. return;
171. end;
172. $$ language plpgsql strict;

三.使用函数装载数据

1. 准备数据


1. select rds_oss_fdw_load_data_prepare('oss_table','lineitem');

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

2. 数据装载


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

函数 rds_oss_fdw_load_data_execute 会等待数据导入的完成才返回。

3. 查询状态

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


1. select application_name, state, pid,query, now() - xact_start as xact  from pg_stat_activity where state != 'idle' and application_name='oss_loader' order by xact desc;

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';

5. 查看进度

我们也很容易看到整个数据装载的进度(单位 MB)


1. select
2. (
3. select sum(size)/1024/1024 as complete from oss_fdw_load_status where status = 1
4. )a,
5. (
6. select sum(size)/1024/1024 as full from oss_fdw_load_status
7. )b;

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，消除了服务器到和客户端的网络开销，有天然的性能优势。
1. dblink 的异步接口非常适合做性能加速，且和过程语言紧密结合。推荐在 SQL 和过程语言中使用。
1. 阿里云开发的 oss_fdw 能在 PostgreSQL 和 OSS 之间做快速的数据交换。oss_fdw 支持 CSV 和压缩方式 CSV 数据的读和写，且很容易用并行加速。oss_fdw 的性能相对于 jdbc insert 和 copy 有压倒的性能优势。

原文：http://mysql.taobao.org/monthly/2018/02/06/