数据库内核月报

数据库内核月报 - 2016 / 12

HybridDB · 最佳实践 · OLAP和OLTP一体化打造

Author: zhibai

HybridDB(基于Greenplum)经过长达四个月时间的公测,终于开始商业化的征程、为我们客户提供计算分析能力。

在这之前,我们团队做了许多技术、产品上的打磨,其中OSS的高效访问与处理是其中较为重要的一环。这个功能可以给用户在数据流转方面带来质的变化。

缘起

在传统的OLAP方案中,链路是比较长的,数据流转的代价较为高昂。而且往往常用的数据同步工具未必能够满足需求,复杂的分析在同步上会需要一些功能的定制。而且资源的不够弹性、管理上的诸多麻烦,也带来成本的上升。

那么在云环境中,这一情况可能发生变化,数据的同步、冷备或许会变得更加简单,在线和离线数据的界线或距离可能会变得不那么明显。

oss_ext 与 oss_fdw

HybridDB和PostgreSQL分别是通过oss_ext和oss_fdw来实现对OSS的读写操作。

注:OSS(Object Storage Service)是阿里云对外提供的海量、安全和高可靠的云存储服务,容量和处理能力的弹性扩展,按实际容量付费,比较适合做数据的冷备。详细请参考对象存储OSS

OSS –> HybridDB 并行加载

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)

关于数据导入,一些最佳实践的建议:

PostgreSQl –> OSS

相比于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 –> OSS –> HybridDB 数据同步

非常重要的一点是: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()

更完善一些的话,可以:

  1. 加上一些错误处理
  2. 加上对OSS的管理部分

数据的分柝与快速分析

这里的“数据的分柝”是指根据原始数据的某个字段或维度,将数据拆分成多份,分开存储与计算。相比要更简单一些,将上面的例子中稍做修改即可。比如,在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实例中,具有相同的表结构的数据,最后需要汇总进行分析、计算,比如分库分表。

那么这个时候,只需将不同的PG实例中的数据写入到同一个OSS目录中、然后在GP中递归加载(prefix参数)上一层目录即可。这里需要注意的有以下几点:

  1. 不同的PG实例中的表结构需要相同,不然GP在加载的时候会报错(某些文件会报字段不存在、或字段多了)
  2. 写入同一个OSS目录中,可以继续分层级,但层级不宜过多

总结

从上面的例子中可以看出:HybridDB + OSS + PostgreSQL,可以有效覆盖OLTP和OLAP一体化的场景。当前在使用上,需要用户做一些自行的定义,同时也提供了足够的灵活度。而在后面,我们可能会进一步进行一体化打造,比如PG到GP的逻辑复制、MySQL到GP的逻辑复制等。让我们在使用体验上更加顺滑。

很荣幸的,见到了很多客户对我们的认可与信赖,甚至有一些客户不断催促我们尽快的商业化,因为他们的业务也要跟着正式上线。这是一份沉甸甸的信任,同时也让我们心存忐忑与敬畏。