Author: zhibai
HybridDB(基于Greenplum)经过长达四个月时间的公测,终于开始商业化的征程、为我们客户提供计算分析能力。
在这之前,我们团队做了许多技术、产品上的打磨,其中OSS的高效访问与处理是其中较为重要的一环。这个功能可以给用户在数据流转方面带来质的变化。
在传统的OLAP方案中,链路是比较长的,数据流转的代价较为高昂。而且往往常用的数据同步工具未必能够满足需求,复杂的分析在同步上会需要一些功能的定制。而且资源的不够弹性、管理上的诸多麻烦,也带来成本的上升。
那么在云环境中,这一情况可能发生变化,数据的同步、冷备或许会变得更加简单,在线和离线数据的界线或距离可能会变得不那么明显。
HybridDB和PostgreSQL分别是通过oss_ext和oss_fdw来实现对OSS的读写操作。
oss_ext
HybridDB是在PostgreSQL的基础上修改,因此和PostgreSQL一样拥有良好、可靠、高效的扩展性。基于HybridDB插件机制,我们开发了oss_ext这个插件,用于在PostgreSQL和HybridDB中无缝访问OSS。
HybridDB的外表功能与PG稍有不同,因此用法上也有细微差别,分为读写两种,详细请参考高速 OSS 并行导入导出。下面我们会以具体例子来介绍。
oss_fdw
PostgreSQL的Foreign Data Wrapper是一套很强大的机制,可以允许用户以SQL方式访问各种各样的数据源,OSS也不在话下。详细的用法参考从 OSS 装载数据到 PostgreSQL
注:OSS(Object Storage Service)是阿里云对外提供的海量、安全和高可靠的云存储服务,容量和处理能力的弹性扩展,按实际容量付费,比较适合做数据的冷备。详细请参考对象存储OSS。
HybridDB是多节点的分布式数据库,在读写OSS的时候,每个节点都会建立一个到OSS的连接,并行读写。如图所示:
数据在导入后,会在每个节点计算HASH值后做分布,相比通过Master的写入,性能可以随节点数线性提升。
create extension oss_ext;
create READABLE external table singlepath (id int, name text) location('oss://oss-cn-beijing.aliyuncs.com dir=loaddata/ id=xxxx key=xxxxx bucket=ext-buffer') format 'csv';
这里的dir参数会指定一个目录loaddata,当前对该表进行查询的时候,会将OSS中该目录下所有文本全部加载进来,但不包括次级目录。执行结果如下:
gpdb=> select * from singlepath;
id | name
----+---------
0 | zero
1 | first
2 | second
(3 rows)
现在在OSS中的目录结构是这样的,两个”output”中的内容是一样的:
loaddata/
-- secondary/
-- output
-- output
创建可以读取层级目录的外部表:
create READABLE external table hyrachi (id int, name text) location('oss://oss-cn-beijing.aliyuncs.com prefix=loaddata/ id=xxxx key=xxxxx bucket=ext-buffer') format 'csv';
查询的结果如下:
gpdb=> select * from hyrachi;
id | name
----+---------
0 | zero
1 | first
2 | second
0 | zero
1 | first
2 | second
(6 rows)
关于数据导入,一些最佳实践的建议:
相比于HybridDB的oss_ext,PostgreSQL的oss_fdw要更强大一些,可以直接进行读写。
执行以下步骤创建相应外表:
create extension oss_fdw ;
CREATE SERVER ossserver FOREIGN DATA WRAPPER oss_fdw OPTIONS (host 'oss-cn-beijing.aliyuncs.com' , id 'xxxx', key 'xxxxx', bucket 'ext-buffer');
create FOREIGN TABLE singpath (id int, name text) SERVER ossserver OPTIONS (dir 'loaddata/', delimiter ',', format 'csv');
查询结果如下:
postgres=> select * from singpath;
id | name
----+---------
0 | zero
1 | first
2 | second
(3 rows)
从另外一张表中写入到OSS外表,从而写入到OSS:
postgres=> select * from oss_examp ;
id | name
----+-------
0 | zero
1 | first
(2 rows)
postgres=> insert into singpath select id, name from oss_examp where id = 0;
NOTICE: begin writiing data to oss directory loaddata/, with block size 32 MB and oss file size 1024 MB
INSERT 0 2
postgres=> select * from singpath ;
NOTICE: a total of 2 files will be loaded
id | name
----+---------
0 | zero
1 | first
2 | second
0 | zero
(5 rows)
利用对OSS外表的读写这一强大的功能,可以很方便地通过SQL实现对数据的清洗、过滤、导出等功能。上面的”insert into … select …” 语句中的select子句可以换成任意条件的查询,从而达到数据处理的目的。
非常重要的一点是:PostgreSQL和HybridDB在接口上基本兼容,基本可以共用。
通过oss_fdw,PostgreSQL可以通过SQL实现很方便地将数据流到OSS中;利用oss_ext,HybridDB则可以通过SQL直接、高效地读取OSS,实现数据的流入。架构如下:
通过在OSS上合理控制目录结构并写入,可以实现PostgreSQL与HybridDB的无缝同步。这里从以下几个例子予以说明:
增量导入一直是被问的比较多的问题,这对于异构数据库一直是一个大的难题。利用oss_ext和oss_fdw则可以基本上通过SQL语句就可以完成需要的工作。
在PG中,建立外表,根据业务需要,定期不定期的根据一定条件将数据同步到OSS的特定目录。例如现在有一张售卖表,每天都有新的售卖记录,需要每天定时统计今天的业务量:
create TABLE sales(id int, name text, deal_day date);
这里是按天来存储数据,一张外表对应OSS中sales目录下的一个子目录,当需要导入新的一天的数据的时候,只需要执行下面两条SQL:
create FOREIGN TABLE sales_2016_12_13 (id int, name text, deal_day date) SERVER ossserver OPTIONS (dir 'sales/2016_12_13/', delimiter ',', format 'csv');
insert into sales_2016_12_13 select * from sales where deal_day == '2016-12-13';
然后,在GP端,可以提前建立好一张总表:
create TABLE sales(id int, name text, deal_day date) distributed by (id);
同样两条SQL,将数据加载进来:
create READABLE external table sales_2016_12_13 (id int, name text) location('oss://oss-cn-beijing.aliyuncs.com dir=sales/2016-12-13/ id=xxxx key=xxxxx bucket=ext-buffer') format 'csv';
insert into sales select * from sales_2016_12_13;
也可以一次性完整的加载:
create READABLE external table sales_ext (id int, name text) location('oss://oss-cn-beijing.aliyuncs.com prefix=sales/ id=xxxx key=xxxxx bucket=ext-buffer') format 'csv';
insert into sales select * from sales_ext;
注:请注意这里的location定义中的”dir” 换成了”prefix”。
将上面的逻辑整理成一个Python脚本就是:
#! /bin/env python
import time
import json
import psycopg2
import datetime
PREFIX = "sales"
def sync_day():
now_date = datetime.datetime.now()
today = "%s_%s_%s" % (now_date.year, now_date.month, now_date.day)
table_name = "%s_%s" % (PREFIX, today)
oss_dir = "%s/%s/" % (PREFIX, today)
print today
print table_name
print oss_dir
pgsql_conn_string = "host=apsaradbsamplepgsqlinstance.pg.rds.aliyuncs.com port=3569 dbname=postgres password= sample user=sample"
pgsql_create_table = "create FOREIGN TABLE %s (id int, name text, deal_day date) SERVER ossserver OPTIONS (dir '%s', delimiter ',', format 'csv');" % ( table_name, oss_dir)
pgsql_insert = "insert into %s select * from %s where deal_day = '%s';" % (table_name, PREFIX, today)
gpdb_conn_string = "host=apsaradbsamplegpdbinstance.gpdb.rds.aliyuncs.com port=3569 dbname=gpdb user=sample password=sample"
gpdb_create_table = "create READABLE external table %s(id int, name text, deal_day date) location('oss://oss-cn-beijing.aliyuncs.com dir=%s id=xxxx key=xxxxx bucket=ext-buffer') format 'csv';" % (table_name, oss_dir)
gpdb_insert = "insert into %s select * from %s where deal_day = '%s';" % (PREFIX, table_name, today)
print pgsql_create_table
execute(pgsql_create_table, pgsql_conn_string)
print pgsql_insert
execute(pgsql_insert, pgsql_conn_string)
print gpdb_create_table
execute(gpdb_create_table, gpdb_conn_string)
print gpdb_insert
execute(gpdb_insert, gpdb_conn_string)
def execute(sql, conn_string):
conn = None
try:
conn = psycopg2.connect(conn_string)
conn.autocommit = True
cur = conn.cursor()
cur.execute(sql)
except Exception as e:
if conn:
try:
conn.close()
except:
pass
time.sleep(10)
print e
return None
def main():
sync_day()
if __name__ == "__main__":
main()
更完善一些的话,可以:
这里的“数据的分柝”是指根据原始数据的某个字段或维度,将数据拆分成多份,分开存储与计算。相比要更简单一些,将上面的例子中稍做修改即可。比如,在PG中,将数据依据条件导出到不同的OSS目录:
create table sales(id int, name text, deal_day date, site_id integer);
create FOREIGN TABLE sales_site_0 (id int, name text, deal_day date, site_id integer) SERVER ossserver OPTIONS (dir 'sales/site_0/', delimiter ',', format 'csv');
create FOREIGN TABLE sales_site_1 (id int, name text, deal_day date, site_id integer) SERVER ossserver OPTIONS (dir 'sales/site_1/', delimiter ',', format 'csv');
create FOREIGN TABLE sales_site_2 (id int, name text, deal_day date, site_id integer) SERVER ossserver OPTIONS (dir 'sales/site_2/', delimiter ',', format 'csv');
insert into sales_site_0 select * from sales where site_id = 0;
insert into sales_site_1 select * from sales where site_id = 1;
insert into sales_site_2 select * from sales where site_id = 2;
导入到OSS中的目录结构是:
sales
--> site_0
--> file1
--> file2
--> file3
--> site_1
--> file1
--> file2
--> site_2
--> file1
--> file2
--> file3
那么,可以将site_0、site_1、site_2分别导入不同的GP实例,或同一个GP实例的不同表中,以达到数据拆分的目的。
当数据量不是特别大的时候,也可以通过下面语句直接进行分析而不必将数据完全导入,以实现快速的检查性分析:
select * from sales_2016_12_13 where deal_day != '2016-12-13';
请注意:这种快速分析的方式,会先将所有数据加载到内存进行计算,不适合大于总内存数据量的场景。
在了解了“增量同步”的部分之后,数据的合并也是比较简单的。这里的“数据合并”是指:多个不同的PG实例中,具有相同的表结构的数据,最后需要汇总进行分析、计算,比如分库分表。
那么这个时候,只需将不同的PG实例中的数据写入到同一个OSS目录中、然后在GP中递归加载(prefix参数)上一层目录即可。这里需要注意的有以下几点:
从上面的例子中可以看出:HybridDB + OSS + PostgreSQL,可以有效覆盖OLTP和OLAP一体化的场景。当前在使用上,需要用户做一些自行的定义,同时也提供了足够的灵活度。而在后面,我们可能会进一步进行一体化打造,比如PG到GP的逻辑复制、MySQL到GP的逻辑复制等。让我们在使用体验上更加顺滑。
很荣幸的,见到了很多客户对我们的认可与信赖,甚至有一些客户不断催促我们尽快的商业化,因为他们的业务也要跟着正式上线。这是一份沉甸甸的信任,同时也让我们心存忐忑与敬畏。