DataBase Foreign Data Wrapper

有时候我们需要将多个数据源的 DataBase 放在一个地方,最 Naive 的方法就是把其他 DataBase 的数据备份出来,再全部导入到一个 DataBase,但是这样比较麻烦,而且当数据库很大时也会比较耗时。这时候使用 FDW 就非常方便了。FDW 全称 Foreign Data Wrapper,这里有一些基本的介绍:Foreign data wrappers - PostgreSQL wiki,FDW 非常简单且效果不错,下面逐步介绍(基于 postgres)基本操作和注意事项。

场景描述

当我们需要从一个 DB(B)获取另一个 DB(A)的表时(常见于各类 BI 需求),一种方法是把 A 的数据完整备份到 B(这里不涉及主从问题),这样做不是不可以,但是比较麻烦;还有一种更为直接的方法就是使用 FDW 插件连接两个 DB,然后就可以直接通过 B 获取 A 的数据。

当然,实际中我们往往还会对从 A 映射过来的数据 Materialized Views 化,简单来说就是对 Table 或 View 进行本地化处理(物理存储)以提升访问速度,唯一需要注意的是需要定期 REFRESH 才能将源 DB 的数据更新,不过这可以通过设置一个定时任务非常简单地实现。

执行步骤

接下来就是这一系列步骤了,包括使用 FDW 映射到 Materialized Views 再到定时任务整个过程的处理。为此,我们事先假设好一个场景,下面的例子将以场景中指定的名字演示。

假设我们有个文本生成的应用:TextGeneration,应用本身只有一张表:text_generation,其他的表如 auth_*django_* 是 Django 自动创建的表,DB 名字为 tg

然后我们还有一个应用:BI,DB 名字为 bi-data,现在想在 BI 的 DB 中使用 TextGeneration 的那张表。

FDW

Step 1: Install the postgres_fdw extension

1
CREATE EXTENSION IF NOT EXISTS postgres_fdw;

Step 2: Create a server

1
2
DROP SERVER IF EXISTS tg_server CASCADE; # prevent duplicated
CREATE SERVER tg_server FOREIGH DATA WRAPPER postgres_fdw OPTIONS (host 'xxx.xxx.xxx.xxx', dbname 'tg', port '5432');

Step3: Create the user mapping

1
CREATE USER MAPPING FOR bi_read_tg_only_user SERVER tg_server OPTIONS (user 'tg_read_only_user', password 'xxxxxxx');

这里需要说明一下如何创建用户:

  • 源 DB 创建为目标 DB 创建一个只读账号,比如 tg_read_only_user
  • 目标 DB 创建一个读写账号,针对从源 DB 映射过来的 schema 只读,或者创建一个只读账号比如 bi_read_tg_only_user
  • 使用目标 DB 的账号去 map 源 DB 为目标 DB 创建的只读账号,即使用 bi_read_only_user 去 map tg_read_only_user
  • 使用目标 DB 的读写或只读账号访问数据,即使用 bi_read_only_user 访问数据

如果两个 DB 在同一个数据库实例里面,那只要注意给不同用户赋权即可。或者简单起见可以只创建一个用户,分别只读源 DB,读写或只读目标 DB

需要特别说明的两点:

  • 有读写权限的用户操作从其他源 DB 映射过来的 schema 时,源 DB 的数据也会发生变化,安全起见建议创建一个只读账号。
  • 如果在目标 DB 中创建了只读用户,默认只有 public 的权限,需要注意给你想要给的 schema 的权限(就是你想读取的从其他 DB 映射过来的 schema)

关于如何创建只读用户以及如何给不同的 schema 权限赋权,可以参考下这篇文章

Step4: Create new schema to access source DB

1
2
DROP SCHEMA IF EXISTS tg_schema; # prevent duplicated
CREATE SCHEMA tg_schema;

Step5: Import schema of source DB

1
2
3
4
IMPORT FOREIGN SCHEMA public FROM SERVER tg_server INTO tg_schema;
# or only import some tables
IMPORT FOREIGN SCHEMA public LIMIT TO ( text_generation )
FROM SERVER tg_server INTO tg_schema;

这里我们第一行命令是把源 DB 的 public schema 导入过来了,你也可以根据需要导入其他 schema。第二行则是只导入指定表格的例子。

到这里我们的 FDW 操作其实已经完成了,此时,你可以在目标 DB 中通过 schema_name.table_name 使用源 DB 的数据啦,注意 table name 是源 DB 的表。

1
SELECT * from tg_schema.text_generation;

Materialized Views

Foreign tables 只是真实 data 的一个代理,为了独立使用这些数据,实际中最简单的复制数据的方法就是使用 materialized views。

Step 1: Create materialized view

1
2
CREATE MATERIALIZED VIEW IF NOT EXISTS tgmv_text_generation AS
SELECT * FROM tg_schema.text_generation; # 这里还可以执行其他 sql

这样就把映射过来的 tg_schema 中的 text_generation 表(也就是源 DB 中的那张表)转为 materialized view 了,我们把 view 命名为 tgmv_text_generation。当然,在 materialized 时可以执行许多其他有用的 SQL 命令,比如 COUNT,GROUP BY 等。

这时就可以直接使用 tgmv_text_generation 了,和上面从 tg_schema.text_generation 中获取的数据应该是一致的。

1
SELECT * from tgmv_text_generation;

那么,当源 DB 的数据发生变化时怎么更新到我们的 materialized views 中呢?这时候就需要进行 REFRESH。

Step2: Refresh data

1
REFRESH MATERIALIZED VIEW tgmv_text_generation;

本来到这里就可以了,但是这里有个 EXCLUSIVE lock 的问题,会导致刷新速度巨慢,详见 Materialized Views 的参考文献。

解决方式是使用 CONCURRENTLY,不过需要定义至少一个 UNIQUE index:

1
2
CREATE UNIQUE INDEX IF NOT EXISTS id ON tgmv_text_generation (id);
REFRESH MATERIALIZED VIEW CONCURRENTLY tgmv_text_generation;

这里的 id 就是源 DB text_generation 表里的 id 列。

批量脚本

实际应用中我们肯定不止一张表格,不可能手动一步一步去执行这些操作,这时候可以通过一个 SQL 命令脚本完成,还以上面的案例为例。

Step 1: Create SQL

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
-- create.sql
-- 映射和赋权可以单独登陆 psql 创建
DO $$
DECLARE r record;
BEGIN
-- 映射
CREATE EXTENSION IF NOT EXISTS postgres_fdw;
DROP SERVER IF EXISTS tg_server CASCADE;
CREATE SERVER tg_server FOREIGH DATA WRAPPER postgres_fdw OPTIONS (host 'xxx.xxx.xxx.xxx', dbname 'tg', port '5432');
CREATE USER MAPPING FOR bi_read_tg_only_user SERVER tg_server OPTIONS (user 'tg_read_only_user', password 'xxxxxxx');
DROP SCHEMA IF EXISTS tg_schema;
CREATE SCHEMA tg_schema;
IMPORT FOREIGN SCHEMA public FROM SERVER tg_server INTO tg_schema;

-- 赋权
GRANT USAGE ON SCHEMA tg_schema TO bi_read_tg_only_user;
GRANT SELECT ON ALL TABLES IN SCHEMA tg_schema TO bi_read_tg_only_user;
-- grant access to the new table automatically
ALTER DEFAULT PRIVILEGES IN SCHEMA tg_schema GRANT SELECT ON TABLES TO bi_read_tg_only_user;

-- 持久化
CREATE EXTENSION IF NOT EXISTS btree_gist;
FOR r IN SELECT tname FROM (VALUES
('auth_group'),
('auth_group_permissions'),
('auth_permission'),
('auth_user'),
('auth_user_groups'),
('auth_user_user_permissions'),
('django_admin_log'),
('django_content_type'),
('django_migrations'),
('django_session'),
('django_site'),
('text_generation')) AS x(tname)
LOOP
-- ignore django_* tables.
IF r.tname like 'django_%' THEN
RAISE NOTICE 'tgmv_% need not to process....', r.tname;
ELSE
EXECUTE format('CREATE MATERIALIZED VIEW IF NOT EXISTS tgmv_%s AS '
'SELECT * FROM tg_schema.%s',
r.tname, r.tname);
EXECUTE format('CREATE UNIQUE INDEX IF NOT EXISTS id ON tgmv_%s (id)',
r.tname);
-- 下面这句是刷新时用的,创建时可以不需要
EXECUTE format('REFRESH MATERIALIZED VIEW CONCURRENTLY tgmv_%s', r.tname);
END IF;
END LOOP;
END$$

然后执行 SQL 命令就可以自动在 DB bi-data 中创建了:

1
$ psql -h [HOST] -p [PORT] -U bi_read_tg_only_user -d bi-data -a -f create.sql

Step 2: Refresh SQL

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
-- refresh.sql
DO $$
DECLARE r record;
BEGIN
FOR r IN SELECT tname FROM (VALUES
('auth_group'),
('auth_group_permissions'),
('auth_permission'),
('auth_user'),
('auth_user_groups'),
('auth_user_user_permissions'),
('django_admin_log'),
('django_content_type'),
('django_migrations'),
('django_session'),
('django_site'),
('text_generation')) AS x(tname)
LOOP
-- ignore django_* tables.
IF r.tname like 'django_%' THEN
RAISE NOTICE 'tgmv_% need not to process....', r.tname;
ELSE
EXECUTE format('REFRESH MATERIALIZED VIEW CONCURRENTLY tgmv_%s', r.tname);
END IF;
END LOOP;
END$$

脚本中我们把所有的表格都放进去了,实际可以只放自己需要的,这样就不需要条件判断了。

定时 REFRESH

实际应用中我们需要创建一个定时执行的脚本任务来批量刷新表格。当然首先推荐的就是 Crontab 了,因为它很简单而且很好用。

Step 1: Create bash script

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# refres.sh
LOGFILE=/tmp/log.txt # absolute path, log file
echo "begin to refresh view..."
echo "=======================" >> $LOGFILE
date +%Y年%m月%d日%H点%M分%S秒 >> $LOGFILE
echo "开始刷新" >> $LOGFILE
echo "=======================" >> $LOGFILE
export PGPASSWORD="xxxxxxxx" # 这里使用了环境变量就不用输密码了
set -e
psql -h [HOST] -p [PORT] -U bi_read_tg_only_user -d bi-data -a -f /absolute/path/to/refresh.sql >> $LOGFILE 2>&1
echo "=======================" >> $LOGFILE
date +%Y年%m月%d日%H点%M分%S秒 >> $LOGFILE
echo "结束刷新" >> $LOGFILE
echo "=======================\n\n" >> $LOGFILE
echo "all views in bi-data have been refreshed..."

Step 2: Create crontab task

1
crontab -e

添加一条命令,比如我们要每天凌晨两点刷新:

1
0 2 * * * /bin/sh /absolute/path/to/refresh.sh

这样每天两点的时候就会自动执行 refresh.sh 并且把日志写入设置好的文件: tmp/log.txt

这一步很有可能会遇到不少坑,最主要的就是定时任务不生效。如果不幸有人遇到了,我建议大家按这里的指导去排查:linux - CronJob not running - Stack Overflow,相信我一定是哪里没搞对,这个东西绝对是可以用的,也要相信自己一定可以搞定。

当然,如果你实在是搞不定或者不喜欢用这个小工具,也可以使用 schedule 起一个独立的服务,专门做数据更新或备份。

参考文献