Digoal.zhou

Digoal.zhou

公益是一辈子的事, I'm digoal, Just do it.

强制数据分布与导出prefix - 阿里云pg, hdb pg oss快速数据规整外部表导出实践案例

2 minute read

背景

批量数据导出在数据分析、数据圈选、广告系统、数据清洗系统、数据规整系统中是一个比较常见的需求，例如：

1、导出任务流。

2、广告系统，圈选人群。

3、数据清洗并导出。

4、数据按某些规则分类，规整并输出到不同的文件。

通常数据库的导出功能相对来说比较单一，例如可以导出query的结果，导出表，导出SCHEMA，导出整个库等。

导出的格式包括CSV，STDOUT等。

但是一个QUERY如果要导出到多个文件（例如前面提到的数据规整需求），目前数据库是没有这样的功能的，需要程序开发。

阿里云RDS PostgreSQL和HDB PostgreSQL提供了这样的导出功能，可以将数据按一定的规则快速的导出到OSS（定义文件的PREFIX，导出的格式等）。

数据按某些规则分类，规整并输出到不同的文件

将数据按某些(字段或虚拟列)内容的不同，写入不同的文件。

这些字段或虚拟列，可以输出到文件，也可以不输出到文件（通过参数调整）。

目前不支持写文件头（列的描述），后面可以加这个功能。或者用户可以在文件最前面加一行，比如用sed或其他编辑工具来添加。

HybridDB for PostgreSQL 例子如下：

1、创建源表

create table t_source (host text, key int , data text);

2、写入一些测试数据

insert into t_source values('host1',1,'1');   
insert into t_source values('host1',1,'2');   
insert into t_source values('host1',2,'3');   
insert into t_source values('host2',1,'4');   
insert into t_source values('host3',1,'5');   
insert into t_source values('host4',1,'6');   
insert into t_source values('host5',1,'7');

3、创建目标表（通用型）

file_name_generator_col，指定这列的内容，作为输出OSS文件的PREFIX。不同的值，输出到不同的文件中。

output_generator_col，指是否将file_name_generator_col列输出到OSS中。false表示不输出。

create WRITABLE external table cdn_demo_output   
(   
  prefix text,  -- 这个列作为分布键、也作为输出文件的PREFIX，同时可以通过output_generator_col参数控制是否将这个列输出到OSS文件中。   
  Data json   
)   
location('@@oss_host@@   
        dir=cdn_demo_20170824/ id=@@oss_id@@   
        key= @@oss_key@@ bucket=@@oss_bucket@@ output_generator_col=false file_name_generator_col=prefix') FORMAT 'csv'    
DISTRIBUTED BY (prefix);

4、创建导出的日志表（元数据表）

create table tbl_output_struct(   
  id int primary key,   -- PK   
  prefix text, -- 前缀   
  struct json, -- 这个前缀对应的，JSON的结构   
  osspath text -- 这个前缀对应的OSS的路径（bucket）。   
);

5、导出到OSS

由于HybridDB for PostgreSQL是分布式数据库，OSS表选择了根据prefix做分布键，所以分布键有几个，就有机会最多往几个SEGMENT去分发，并行写OSS。如果只有一个prefix值，那么就只会分发到一个SEGMENT写OSS。

用户可以根据需要，选择几个PREFIX，当然PREFIX还可以在分词prefix1和prefix2组成的联合prefix。

例如：

国家||'_'||省份     
这个相当于两个字段联合的prefix。   
   
   
taskid||'_'||(random()*9)::int::text   
从而每个taskid可以拆成10个prefix2，相当于一个taskid开了10个并行写。   
（当没有第二个字段时，使用这种手段效果一样可以多个节点并行）

导出：

begin;   
   
-- 记录下此次导出的结构，PREFIX等(若已知)。   
insert into tbl_output_struct values ('label1', '{col:type1, col2:type2, ....}', 'bucket_path');   
-- OR   
insert into tbl_output_struct select host||key, '{col:type1, col2:type2, ....}', 'bucket_path' from t_source group by 1;   
   
-- 打开按prefix导出的参数开关   
set rds_write_oss_file_by_distribution_column=on;   
   
-- 导出到OSS   
insert into cdn_demo_output   
(   
  prefix,   
  Data   
)   
select    
  host_and_key,                       -- prefix驱动键   
  row_to_json(row(host_and_key,data)  -- 将需要导出的内容，封装到JSON中   
from   
(   
select    
  row_number() over (partition by host||key order by host||key) as RN,     -- 强制数据库按PREFIX的顺序排序，这样才能保证prefix写入到对应的文件   
  host||key as host_and_key,     
  data    
  from t_source t1    
) t;   
   
end;

PS，强制分布的原理：

                                                  QUERY PLAN                                                     
--------------------------------------------------------------------------------------------------------------   
 Gather Motion 48:1  (slice2; segments: 48)  (cost=0.04..0.06 rows=1 width=64)   
   ->  Subquery Scan t  (cost=0.04..0.06 rows=1 width=64)   
         ->  Window  (cost=0.04..0.05 rows=1 width=68)   
               Partition By: host_and_key   
               Order By: host_and_key   
               ->  Sort  (cost=0.04..0.04 rows=1 width=68)   
                     Sort Key: host_and_key   
                     ->  Redistribute Motion 48:48  (slice1; segments: 48)  (cost=0.00..0.03 rows=1 width=68)   
                           Hash Key: host_and_key   
                           ->  Seq Scan on t_source t1  (cost=0.00..0.01 rows=1 width=68)   
 Settings:  enable_bitmapscan=off; enable_seqscan=off; optimizer=off   
 Optimizer status: legacy query optimizer   
(12 rows)

6、格式转换

导出为JSON格式，如果需要转换为CSV，用户可以将数据从OSS读出之后，写程序转换为CSV格式。

格式的内容在元数据表里面。

OSS将来应该也会提供这样的函数编程接口，对OSS里面的数据进行格式转换。

7、将结果作为OSS外部表

将导出的内容，再提供查询。

由于前面我们设置了output_generator_col=false，所以不输出分布键，那么实际上在OSS文件中，只有一列，所以只读OSS外部表的定义如下

create external table cdn_demo_check   
(   
Data json   
)   
location('@@oss_host@@   
        dir=cdn_demo_20170824/ id=@@oss_id@@   
        key= @@oss_key@@ bucket=@@oss_bucket@@') FORMAT 'csv';   
   
select * from  cdn_demo_check order by data;

其他 强制输出到N个OSS文件

如果用户想把结果集所有内容都输出到一个OSS文件怎么办呢？

加一个伪劣，输出单一值，并在输出前按这个伪劣来强制分布即可。

https://yq.aliyun.com/articles/582803

使用HybridDB PG的外部表输出数据到OSS时，一般会输出成多个文件（文件个数一般与节点数个数一致）。如何输出为一个文件呢？步骤如下：

1）创建示例表：

create table t3 (a1 text, a2 text, a3 text, a4 text) distributed by (a1);  
  
insert into t3 values('xxxxxxx','yyyyy','zzzzz','wwwww');  
insert into t3 select * from test;  
insert into t3 select * from test;  
insert into t3 select * from test;  
insert into t3 select * from test;  
insert into t3 select * from test;  
insert into t3 select * from test;

2）创建外部表写表，按源数据表的结构创建外部表。注意：

  • 相对于本地源表的表结构，需要添加一个临时字段，例如dummy_col，最好作为第一个字段。

  • 添加两个选项: output_generator_col=false 与 file_name_generator_col=dummy_col

    file_name_generator_col 这个参数指定表中添加的字段。

drop external table test_oss_write;  
create writable external table test_oss_write(  
  dummy_col text,  
  A1  text,  
  A2  text,  
  A3  text,  
  A4  text  
)  
location('oss://oss-cn-shanghai.aliyuncs.com   
dir=output_x_file/  
id=xxx  
key=xxx  
bucket=osshuadong2   
output_generator_col=false   
file_name_generator_col=dummy_col  
') FORMAT 'csv'  ( DELIMITER ',')   
distributed by (dummy_col)  
;

3) 定制 SQL 将test数据写入到 oss 中

set rds_write_oss_file_by_distribution_column=on;   
  
insert into test_oss_write  
Select key,a1,a2,a3,a4 from  
(  
select  
  floor(random()*(5-1)+1)::int::text || 'e' as key,  
  A1,A2,A3,A4     
  from t3  
)t order by key;

注意：

  • 使用随机数来指定产生的文件个数，这里我们需要输出4个文件，则使用 floor(random()*(5-1)+1)

  • 写入文件时，给到写入节点的数据需要按照key有序，我们使用了order by, 也可以使用窗口函数。

最后OSS上能看到4个文件，且文件内容不包含用来定制文件名的虚拟列

$osscmd ls  oss://osshuadong2/output_x_file  
  
prefix list is:   
object list is:   
2018-04-18 15:54:18 2.46KB Standard oss://osshuadong2/output_x_file/1e_577353258534704.1  
2018-04-18 15:54:18 2.46KB Standard oss://osshuadong2/output_x_file/2e_577353258534704.2  
2018-04-18 15:54:18 2.51KB Standard oss://osshuadong2/output_x_file/3e_577353258534704.2  
2018-04-18 15:54:18 2.34KB Standard oss://osshuadong2/output_x_file/4e_577353258534704.1  
  
osscmd get oss://osshuadong2/output_x_file/3e_577353258534704.2 b.txt  
  
cat b.txt   
  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww  
xxxxxxx,yyyyy,zzzzz,wwwww

参考

RDS PG OSS 外部表文档：https://help.aliyun.com/knowledge_detail/43352.html

HDB PG OSS 外部表文档：https://help.aliyun.com/document_detail/35457.html

《日增量万亿+级 实时分析、数据规整 - 阿里云HybridDB for PostgreSQL最佳实践》

Flag Counter

digoal’s 大量PostgreSQL文章入口

You May Also Enjoy

PostgreSQL(PPAS 兼容Oracle) 从零开始入门手册 - 珍藏版

17 minute read

背景 云数据库PPAS版，是阿里云与EnterpriseDB公司(简称EDB)合作基于PostgreSQL高度兼容Oracle语法的数据库服务，为用户提供易于操作的迁移工具，兼容范围涵盖：PL/SQL、数据类型、高级函数、表分区等。 用户可以直接在阿里云购买PPAS进行使用。 如果在购买PPAS前，想试用一下...

PostgreSQL pipelinedb 流计算插件 - IoT应用 - 实时轨迹聚合

1 minute read

背景 IoT场景，车联网场景，共享单车场景，人的行为位点等，终端实时上报的是孤立的位点，我们需要将其补齐成轨迹。 例如共享单车，下单，开锁，生成订单，骑行，关闭订单，关锁。这个过程有一个唯一的订单号，每次上报的位点会包含时间，订单号，位置。 根据订单号，将点聚合为轨迹。 使用pipelinedb插件，可以实...

PostgreSQL datediff 日期间隔（单位转换）兼容SQL用法

2 minute read

背景 使用datediff，对时间或日期相减，得到的间隔，转换为目标单位(日、月、季度、年、小时、秒。。。等)的数值。 DATEDIFF ( datepart, {date|timestamp}, {date|timestamp} ) 周 select datediff(week,'2009-01-0...