Greenplum PXF使用与思考

一、背景说明

        上周支撑某现场时接触到了greenplum，本文用来记录greenplum的一些常用使用命令，以及个人对Greenplum的一些思考。

        我司的大数据平台通过JDBC、PXF两种方式将外部数据源导入到greenplum组件。JDBC方式，即和Spark往oracle等关系型数据库一样，直接调用jdbc()方法，触发action即可。PXF方式导入数据分为以下几步：

1. 将外部数据先写成 Avro/JSON/RCFile/Parquet/SequenceFile/ORC/Text格式，我司平台选择的是parquet文件。
2. 创建GP外表，关联上一步中落地的文件。
3. 将外表的数据用select insert语句导入到GP内表。

二、greenplum命令

1、登录gp

# -d,指定数据库;-U指定登录用户;
psql -d postgres -U daedb

2、查看集群中部署了多少segment节点

select * from gp_segment_configuration;

3、查看表数据在segment上的分布情况，即数据是否倾斜

select gp_segment_id,count(1) from test55 group by gp_segment_id;

4、查看版本

# 输出结果中包括postgresql和gp版本;
select version();

5、查询GP 正在执行的SQL

select usename,query_start,current_query,client_addr from pg_stat_activity where current_query!='<IDLE>';

6、创建GP外表，关联hdfs上的parquet文件

# 此外表只能用于读,不能继续写入数据;(我司的平台中,外表算是一个中间存储结果,建完外表后不需要再往里面写数据)
create external table  nj.AD15_external("AAE060" decimal(38,10),"AAE076" decimal(38,10),"AAE140" text) LOCATION ('pxf://hdfsOutput/audit-caiwu/ad15/2562472484057065?PROFILE=parquet') FORMAT 'CUSTOM' ( FORMATTER='pxfwritable_import');

可写入的外表，建表语句可参考

CREATE [WRITABLE] EXTERNAL TABLE <table_name>
    ( <column_name> <data_type> [, ...] | LIKE <other_table> )
LOCATION ('pxf://<path-to-hdfs-dir>
    ?PROFILE=hdfs:parquet')
FORMAT 'CUSTOM' (FORMATTER='pxfwritable_export');
[DISTRIBUTED BY (<column_name> [, ... ] ) | DISTRIBUTED RANDOMLY];

三、对GP PXF的思考

下图是greenplum组件的架构图：

1. 每个segment节点上，都要部署PXF服务。PXF通过REST服务和本机的segment服务交互。（GP集群分master节点、segment节点）
2. PXF根据pxf-profiles.xml里配置的fragmenter“去切分“数据文件，根据accessor、resolver去解析文件具体的字段值。

例如配置pxf可以去解析hdfs上的parquet文件格式：

<!--该配置样例在github greenplum-db/pxf项目上可以找到-->
<profile>
        <name>Parquet</name>
        <description>A profile for reading and writing Parquet data from HDFS</description>
        <plugins>
            <fragmenter>org.greenplum.pxf.plugins.hdfs.HdfsDataFragmenter</fragmenter>
            <accessor>org.greenplum.pxf.plugins.hdfs.ParquetFileAccessor</accessor>
            <resolver>org.greenplum.pxf.plugins.hdfs.ParquetResolver</resolver>
        </plugins>
</profile>

1、PXF拆分hdfs parquet文件拆分机制

protected List<InputSplit> getSplits(Path path) throws IOException {
     // PxfInputFormat,
        PxfInputFormat pxfInputFormat = new PxfInputFormat();
        PxfInputFormat.setInputPaths(jobConf, path);
        InputSplit[] splits = pxfInputFormat.getSplits(jobConf, 1);
        // 省略部分代码;
        return splits;
}

其中PxfInputFormat继承了FileInputFormat

public class PxfInputFormat extends FileInputFormat

FileInputFormat类中定义了InputSplit切分的规则，其中split size计算规则如下：

// 当文件是可拆分的情况下,spark/MapReduce拆分任务的大小计算规则
// minSize,mapreduce.input.fileinputformat.split.minsize,默认1L; 
// maxSize,mapreduce.input.fileinputformat.split.maxsize,默认Long的大值;
// 因此,默认情况下split size就是blocksize，即128MB;
protected long computeSplitSize(long blockSize, long minSize,
                                  long maxSize) {
    return Math.max(minSize, Math.min(maxSize, blockSize));
}

从这边可以看出，默认情况下greenplum pxf拆分“任务”是根据parquet rowgroup大小来拆分。

四、入库性能

        现场yarn节点共6台，gp集群共6台机器，Spark读取Oracle数据时按照rownum来划分任务，在入库时每个任务处理200w数据，pxf方式下入库数据如下：

从上述表格可以看出：

• Oracle表数据量上亿后，入库速度明显下降。
• 在现场处理Oracle 20+亿的数据，Spark写parquet文件就会消耗很长的时间（没有等到写完就放弃了，因此Oracle表数据量上亿时，起码要有分区，控制分区内的数据量）

五、其它

1、greenplum pxf源代码地址

# 老版本的pxf项目地址;
https://github.com/apache/incubator-hawq
# 新版本的pxf项目地址;
https://github.com/greenplum-db/pxf