助力Greenplum数据库开发之接口篇上GolangCPython和Perl

35次阅读

共计 15800 个字符，预计需要花费 40 分钟才能阅读完成。

Greenplum 作为一款强大的 HTAP 数据库，针对大多数流行语言都有相应的连接库。大部分均是与 PostgreSQL 采用相同的接口，但是也有部分接口是 Greenplum 专门优化后用于自身使用的。今天，我们将给大家分享一系列语言接口的连接库及使用方式。

1. Golang

Golang 作为 Google 开源的一款编译型开发语言，经过多年发展，在开发界占据了很大的份额，市面上针对 Greenplum 的 Golang 连接库也有很多，但是最著名的还是 github.com/lib/pq。

Golang 为连接数据库专门提供了一个公共包叫 database/sql，在进行数据库连接时，只需要遵循该库的注册方式将 lib/pq（https://github.com/lib/pq）注册为 postgres 驱动即可。

支持列表

目前该包支持全系列 Greenplum 产品，提供的一些功能如下：

  • 支持数据库增删改查
  • 支持在数据库中直接执行 DDL 或维护语句（VACUUM 等）
  • 支持所有普通数据类型
  • 支持 bytea
  • 支持 hstore
  • 支持 SSL
  • 支持 COPY FROM
  • 支持 pgpass

安装方法

与其他的 Go 包安装方式一样，只需要在 go/src 目录下执行 go get 命令即可将需要的包下载。go get github.com/lib/pq 如果网络条件有限，不能直接 go get 联网下载，也可以直接在外网机器上用 go get 或者 git 下载好，然后将代码复制到对应格式的目录下（放到 go 文件夹下的 src 文件夹下的 github.com 文件夹下的 lib 文件夹下的 pq 下）。

使用方法

下面展示一段 Golang 连接数据库查询的代码，进行简单分析。

package main

import (
    "database/sql"
    "fmt"
    _ "github.com/lib/pq"
    "log"
)

func main() {db, err := sql.Open("postgres", "user=chris password=123 dbname=postgres host=127.0.0.1 port=5432 sslmode=disable")
    if err != nil {log.Fatal(err)
    }
    defer db.Close()

    // 查询数据
    rows, err := db.Query("select version()")

    for rows.Next() {
        var version string
        rows.Scan(&version)
        fmt.Println(version)
    }
}

从上面代码可以看到，我们需要首先导入 database/sql 和 lib/pq 两个包。

import (“database/sql””fmt”_ “github.com/lib/pq””log”)注意这里在导入 lib/pq 时，前面加了下划线（_），因为通常来说，不直接使用驱动所提供的方法，而是应该使用 database 中的 sql.DB，因此在导入 lib/pq 驱动时，这里使用了匿名导入的方式(在包路径前添加 _)，当导入了一个数据库驱动后，此驱动会自行初始化并注册自己到 Golang 的 database/sql 上下文中，因此我们就可以通过 database/sql 包提供的方法访问数据库了。

database/sql 默认提供了 MySQL、PostgreSQL 和 SQLite 的支持，不需要手工注册。

接下来就是在函数中构造连接 url 然后进行查询了

以下就是具体的连接使用逻辑，通常的逻辑一般为：

  • 打开数据库连接
  • 执行增删改查操作
  • 扫描结果集然后处理（查询操作）
  • 错误处理
  • 关闭数据库连接 数据源的连接字符串通常拼接成“user=chris password=123
    dbname=postgres host=127.0.0.1 port=5432 sslmode=disable”的形式，字符串中分别对应了 Greenplum 数据库的用户名、密码、数据库名、主机 ip、Greenplum 端口号和 sslmode。

db, err := sql.Open(“postgres”, “user=chris password=123 dbname=postgres host=127.0.0.1 port=5432 sslmode=disable”)
除了上面的数据源连接串形式，也可以下面这种方式连接（不太常用，作用是一样的）：“postgres://chris:password@127.0.0.1:5432/postgres?sslmode=verify-full”。

数据库的操作支持开发语言中常见的两种处理方式：

  1. 直接查询
  2. prepare statement

下面给大家展示所有可能涉及的增删改查操作的语法，可以直接在数据库中测试使用。

package main

import (
    "database/sql"
    "fmt"
    _ "github.com/lib/pq"
    "time"
)

var db *sql.DB

func sqlOpen() {
    var err error
    db, err = sql.Open("postgres", "user=gpadmin password=123 dbname=postgres host=172.16.142.191 port=5432 sslmode=disable")
    //port 是数据库的端口号，默认是 5432，如果改了，这里一定要自定义；//user 就是你数据库的登录帐号;
    //dbname 就是你在数据库里面建立的数据库的名字;
    //sslmode 就是安全验证模式;
    checkErr(err)

}

func sqlCreate() {
    // 创建表
    _, err := db.Exec("drop table t_user")
    checkErr(err)
    _, err1 := db.Exec("create table t_user(uname text,dptname text,create_time timestamp)")
    checkErr(err1)
}

func sqlInsert() {
    // 插入数据
    stmt, err := db.Prepare("INSERT INTO t_user(uname,dptname,create_time) VALUES($1,$2,$3)")
    checkErr(err)

    _, err = stmt.Exec("chris", "软件 1 部", "2020-01-08")
    // 这里的三个参数就是对应上面的 $1,$2,$3 了

    checkErr(err)
}

func sqlDelete() {
    // 删除数据
    stmt, err := db.Prepare("delete from t_user where uname=$1")
    checkErr(err)

    res, err := stmt.Exec("chris")
    checkErr(err)

    affect, err := res.RowsAffected()
    checkErr(err)

    fmt.Println("rows affect:", affect)
}

func sqlSelect() {
    // 查询数据
    rows, err := db.Query("SELECT * FROM t_user")
    checkErr(err)

    println("-----------")
    for rows.Next() {
        var uname string
        var dptname string
        var create_time string
        err = rows.Scan(&uname, &dptname, &create_time)
        checkErr(err)
        fmt.Println("name =", uname, "\ndep =", dptname, "\ncreated =", create_time, "\n")
    }
}

func sqlUpdate() {
    // 更新数据
    stmt, err := db.Prepare("update t_user set dptname=$1 where uname=$2")
    checkErr(err)

    res, err := stmt.Exec("软件 1 部","jenny")
    checkErr(err)

    affect, err := res.RowsAffected()
    checkErr(err)

    fmt.Println("rows affect:", affect)
}
func sqlClose() {db.Close()
}

func checkErr(err error) {
    if err != nil {panic(err)
    }
}

func main() {
    sep := "----------\n"
    sqlOpen()
    println(sep, "*sqlOpen")

    sqlCreate()
    println(sep, "*sqlCreate")
    time.Sleep(time.Second*2)

    sqlSelect()
    println(sep, "*sqlSelect")
    time.Sleep(time.Second*2)

    sqlInsert()
    sqlSelect()
    println(sep, "*sqlInsert")
    time.Sleep(time.Second*5)

    sqlUpdate()
    sqlSelect()
    println(sep, "*sqlUpdate")
    time.Sleep(time.Second*2)

    sqlDelete()
    sqlSelect()
    println(sep, "*sqlDelete")
    time.Sleep(time.Second*2)

    sqlClose()
    println(sep, "*sqlClose")
}

2. C 语言

C 语言是比较偏基础的语言，现在大多数开发人员可能不怎么会用 C 语言进行开发。但是在 2020 年 1 月份发布的 2019 年度编程语言（
https://www.tiobe.com/tiobe-i…）中，C 语言战胜众多对手，王者归来，足见 C 语言的地位。

也确实如上所说，C 语言可能影响到软件开发的方方面面，但并不一定为你我所熟知。在 PostgreSQL/Greenplum 领域中，libpq 是 PostgreSQL/Greenplum 的 C 应用程序接口，这个 C 接口同时也是 C++、Perl、Python、Tcl 和 ECPG 应用接口的驱动引擎。今天就来简单看看这个 C 接口是如何使用的。

示例

可以将如下代码编辑到 gpadmin 用户下的 testlibpq.c 文件中，下面会介绍简单的编译和使用

/*
 * src/test/examples/testlibpq.c
 * testlibpq.c
 *      注意：下面需要引入 libpq-fe.h 头文件
 */
#include <stdio.h>
#include <stdlib.h>
#include "libpq-fe.h"

static void
exit_nicely(PGconn *conn)
{PQfinish(conn);
    exit(1);
}

int
main(int argc, char **argv)
{
    const char *conninfo;
    PGconn     *conn;
    PGresult   *res;
    int         nFields;
    int         i,
                j;

    /* 
     * 如果用户在命令行定义了参数，那么这里解析为 conninfo，作为连接参数；* 如果没有定义，则会选用环境变量定义的连接参数或采用默认值。*/
    if (argc > 1)
        conninfo = argv[1];
    else
        conninfo = "dbname = postgres";

    /* 创建一个数据库连接 */
    conn = PQconnectdb(conninfo);

    /* 检查连接是否成功开启 */
    if (PQstatus(conn) != CONNECTION_OK)
    {
        fprintf(stderr, "Connection to database failed: %s",
                PQerrorMessage(conn));
        exit_nicely(conn);
    }

    /* Set always-secure search path, so malicious users can't take control. */
    res = PQexec(conn,
                 "SELECT pg_catalog.set_config('search_path','', false)");
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {fprintf(stderr, "SET failed: %s", PQerrorMessage(conn));
        PQclear(res);
        exit_nicely(conn);
    }

    /*
     * 当完成操作后，执行清理操作，避免内存泄漏。*/
    PQclear(res);

    /*
     * Our test case here involves using a cursor, for which we must be inside
     * a transaction block.  We could do the whole thing with a single
     * PQexec() of "select * from pg_database", but that's too trivial to make
     * a good example.
     */

    /* 开始一个事务块 */
    res = PQexec(conn, "BEGIN");
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
    {fprintf(stderr, "BEGIN command failed: %s", PQerrorMessage(conn));
        PQclear(res);
        exit_nicely(conn);
    }
    PQclear(res);

    /*
     * 从 pg_database 表中获取数据
     */
    res = PQexec(conn, "DECLARE myportal CURSOR FOR select * from pg_database");
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
    {fprintf(stderr, "DECLARE CURSOR failed: %s", PQerrorMessage(conn));
        PQclear(res);
        exit_nicely(conn);
    }
    PQclear(res);

    res = PQexec(conn, "FETCH ALL in myportal");
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {fprintf(stderr, "FETCH ALL failed: %s", PQerrorMessage(conn));
        PQclear(res);
        exit_nicely(conn);
    }

    /* 打印列名 */
    nFields = PQnfields(res);
    for (i = 0; i < nFields; i++)
        printf("%-15s", PQfname(res, i));
    printf("\n\n");

    /* 打印数据 */
    for (i = 0; i < PQntuples(res); i++)
    {for (j = 0; j < nFields; j++)
            printf("%-15s", PQgetvalue(res, i, j));
        printf("\n");
    }

    PQclear(res);

    /* 关闭上面定义的 cursor */
    res = PQexec(conn, "CLOSE myportal");
    PQclear(res);

    /* end the transaction */
    res = PQexec(conn, "END");
    PQclear(res);

    /* 关闭数据库连接 */
    PQfinish(conn);

    return 0;
}

编译

[gpadmin@gp1 ~]$ gcc -I /usr/local/greenplum-db/include/ -L /usr/local/greenplum-db/lib -lpq testlibpq.c -o testlibpq
这里采用 gcc 编译器，指定头文件 (-I) 和库文件(-L)，编译完成后，会生成二进制可执行文件 testlibpq。

「实战」助力数据库开发之接口篇 – C 连接 Greenplum
运行 此时可以执行该文件，查看是否能正常访问数据库。

[gpadmin@gp1 ~]$ ./testlibpq
datname datdba encoding datcollate datctype datistemplate datallowconn datconnlimit datlastsysoid datfrozenxid datminmxid dattablespace datacl

template1 10 6 en_US.utf8 en_US.utf8 t t -1 12813 725 1 1663 {=c/gpadmin,gpadmin=CTc/gpadmin}
template0 10 6 en_US.utf8 en_US.utf8 t f -1 12813 725 1 1663 {=c/gpadmin,gpadmin=CTc/gpadmin}
postgres 10 6 en_US.utf8 en_US.utf8 t t -1 12813 725 1 1663
上面代码采用默认的连接信息，如果要特别指定，可以通过定义如下字符串信息来访问数据库。

[gpadmin@gp1 ~]$ ./testlibpq “host=172.16.142.191 port=5432 user=gpadmin dbname=postgres”
datname datdba encoding datcollate datctype datistemplate datallowconn datconnlimit datlastsysoid datfrozenxid datminmxid dattablespace datacl

template1 10 6 en_US.utf8 en_US.utf8 t t -1 12813 725 1 1663 {=c/gpadmin,gpadmin=CTc/gpadmin}
template0 10 6 en_US.utf8 en_US.utf8 t f -1 12813 725 1 1663 {=c/gpadmin,gpadmin=CTc/gpadmin}
postgres 10 6 en_US.utf8 en_US.utf8 t t -1 12813 725 1 1663
在尝试连接过程中，可能会报如下错误，此时需要修改 pg_hba.conf 文件并 gpstop -u 生效即可。

[gpadmin@gp1 ~]$ ./testlibpq
Connection to database failed: FATAL: no pg_hba.conf entry for host “[local]”, user “gpadmin”, database “postgres”, SSL off

参考文章

[1] https://www.postgresql.org/do…

3. Python

现在我们再来看一下，在这几年火的一塌糊涂的 Python 语言如何连接 Greenplum。

Python 连接 Greenplum 数据库较常用的库有 PyGreSQL 和 Psycopg2 两个。Greenplum 的很多脚本都是采用 PyGreSQL 为基础开发的，可见 PyGreSQL 肯定有其独到之处，但是 Psycopg2 这几年似乎在 Postgres 体系中更加流行。本文将会分别介绍这两个库的使用。

PyGreSQL

PyGreSQL 是连接 PostgreSQ L 的 Python 库，目前最新版本为 PyGreSQL 5.1，支持 PostgreSQL 9.0 到 11 版本，可以对应到 Greenplum 6.x 的版本，如果要支持 Greenplum 4.x 和 5.x 版本，可以选用 PyGreSQL 4.x 版本。

安装

pip install PyGreSQL

示例

#!/usr/bin/env python

import pg


def operate_postgre_tbl_product():
    try:
        #pgdb_conn = pg.connect(dbname = 'tpc', host = '192.168.103.31', user = 'gpadmin', passwd = '')
        pgdb_conn = pg.connect("host=192.168.103.31 port=5432 dbname=tpc user=gpadmin")

    except Exception, e:
         print e.args[0]
         return


    sql_desc = "select * from call_center limit 5;"
    for row in pgdb_conn.query(sql_desc).dictresult():
        print row


    pgdb_conn.close()


if __name__ == '__main__':
    operate_postgre_tbl_product()

参考文章

http://www.pygresql.org/about…

Psycopg2

Psycopg2 库的底层是由 C 语言封装 PostgreSQL 的标准库 C 接口库 libpq 实现的，运行速度非常快，它支持大型多线程应用的大量并发 Insert 和 Update 操作，另外它完全兼容 DB API 2.0。

安装

pip install psycopg2

示例

1、简单的增加，查询记录

import psycopg2
import psycopg2.extras
import time
 
'''
    连接数据库
    returns:db
'''
def gp_connect():
    try:
        db = psycopg2.connect(dbname="testdb",
                              user="gpadmin",
                              password="gpadmin",
                              host="10.1.208.42",
                              port="5432")
        # connect()也可以使用一个大的字符串参数,
        # 比如”host=localhost port=5432 user=postgres password=postgres dbname=test”return db
    except psycopg2.DatabaseError as e:
        print("could not connect to Greenplum server",e)
 
 
if __name__ == '__main__':
    conn = gp_connect()
    print(conn)
    cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
    # 这里创建的是一个字典 Cursor, 这样返回的数据, 都是字典的形式, 方便使用
    ret = cur.execute("CREATE TABLE public.gp_test (id serial PRIMARY KEY, num integer, data varchar);")
    conn.commit()
    # 提交到数据库中
    print(ret)
    ret = cur.execute("INSERT INTO public.gp_test (num, data) VALUES (%s, %s);",(300, "abc'def"))
 
    conn.commit()
    # 提交到数据库中
    print(cur.rowcount)  # 1
    # 返回数据库中的行的总数已修改，插入或删除最后 execute*().
 
    ret_sql = cur.mogrify("select * from pg_tables where tablename = %s;", ('gp_test',))
    # 返回生成的 sql 脚本, 用以查看生成的 sql 是否正确.
    # sql 脚本必须以; 结尾, 不可以省略. 其次, 不管 sql 中有几个参数, 都需要用 % s 代替, 只有 % s, 不管值是字符还是数字, 一律 % s.
    # 最后, 第二个参数中, 一定要传入元组, 哪怕只有一个元素, 像我刚才的例子一样, ('gp_test')这样是不行的.
    print(ret_sql.decode('utf-8'))  # select * from pg_tables where tablename = E'gp_test';
 
    cur.execute("select * from gp_test where num = %s;", (300,))
    pg_obj = cur.fetchone()
    print(pg_obj) # {'id': 1, 'num': 300, 'data': "abc'def"}
 
    conn.close() # 关闭连接

2、批量插入，查询

conn = gp_connect()
print(conn)
cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
# # 这里创建的是一个字典 Cursor, 这样返回的数据, 都是字典的形式, 方便使用
# ret = cur.execute("CREATE TABLE public.gp_test (id serial PRIMARY KEY, num integer, data varchar);")
# conn.commit()
# # 提交到数据库中
# print(ret)
gp_list = []
for i in range(200):
    gp_list.append((i,'abc%s'%i))
# print(gp_list)
# 批量提交数据
ret = cur.executemany("INSERT INTO public.gp_test (num, data) VALUES (%s, %s);", gp_list)
conn.commit()
# 提交到数据库中
print(cur.query)  # 查看上一条执行的脚本
print(cur.rowcount)  # 200
# 返回数据库中的行的总数已修改，插入或删除最后 execute*().
cur.execute("select  count(*) num from gp_test")
pg_obj = cur.fetchone()
print(pg_obj)  # {'num': 200}
 
conn.close()  # 关闭连接

3、使用连接池，执行高性能的批量插入与查询

import psycopg2
import psycopg2.extras
import psycopg2.pool
from datetime import datetime
 
'''
    连接数据库
    使用数据库连接池
    returns:db
'''
def gp_connect():
    try:
        simple_conn_pool = psycopg2.pool.SimpleConnectionPool(minconn=1, maxconn=5,dbname="testdb",
                              user="gpadmin",
                              password="gpadmin",
                              host="10.1.208.42",
                              port="5432")
        # connect()也可以使用一个大的字符串参数,
        # 比如”host=localhost port=5432 user=postgres password=postgres dbname=test”# 从数据库连接池获取连接
        conn = simple_conn_pool.getconn()
        return conn
    except psycopg2.DatabaseError as e:
        print("could not connect to Greenplum server",e)
 
 
if __name__ == '__main__':
    conn = gp_connect()
    print(conn)
    cur = conn.cursor()
    # 批量查询大小
    batch_size = 1000
    gp_list = []
    for i in range(2000, 100000):
        gp_list.append((i,'abc%s'%i))
    # print(gp_list)
 
    # 开始时间
    start_time = datetime.now()
    # 批量提交数据 execute_values 性能大于 executemany
    psycopg2.extras.execute_values(cur, "INSERT INTO public.gp_test (num, data) VALUES %s", gp_list)
    conn.commit()
    # 提交到数据库中
    cur.execute("select  *  from gp_test order by id")
    count = 0
 
    while True:
        count = count + 1
        # 每次获取时会从上次游标的位置开始移动 size 个位置，返回 size 条数据
        data = cur.fetchmany(batch_size)
        # 数据为空的时候中断循环
        if not data:
            break
        else:
            print(data[-1])  # 得到最后一条(通过元祖方式返回)
        print('获取 %s 到 %s 数据成功' % ((count - 1) * batch_size, count * batch_size))
    print('insert 到 fetchmany 获取全量数据所用时间:', (datetime.now() - start_time).seconds) # 16s
conn.close()  # 关闭连接

4、执行高性能的批量更新与查询

import psycopg2
import psycopg2.extras
import psycopg2.pool
from datetime import datetime

'''
    连接数据库
    使用数据库连接池
    returns:db
'''
def gp_connect():
    ……略

if __name__ == '__main__':
    conn = gp_connect()
    print(conn)
    cur = conn.cursor()
    # 批量查询大小
    batch_size = 1000
    gp_uplist = [] # 更新列表
    for i in range(2000, 10000):
        gp_uplist.append((i,'def%s'%i))
    print(gp_uplist)

    # 开始时间
    start_time = datetime.now()
    # 批量提交数据 execute_values 性能大于 executemany

    sql = "UPDATE public.gp_test SET data = TEST.data" \
          "FROM (VALUES %s) AS TEST(num, data)" \
          "WHERE public.gp_test.num = TEST.num"
    # 批量更新语句模版 UPDATE TABLE SET TABLE.COL = XX.col
    # FROM (VALUES %s) AS XX(id_col,col)
    # WHERE TABLE.id_col = XX.id_col 
    # XX 为别名
    psycopg2.extras.execute_values(cur, sql, gp_uplist, page_size=100)
    print(cur.query)
    conn.commit()
    # 提交到数据库中
    cur.execute("select  *  from gp_test order by id")
    count = 0

    while True:
        count = count + 1
        # 每次获取时会从上次游标的位置开始移动 size 个位置，返回 size 条数据
        data = cur.fetchmany(batch_size)
        # 数据为空的时候中断循环
        if not data:
            break
        else:
            print(data[-1])  # 得到最后一条(通过元祖方式返回)
        print('获取 %s 到 %s 数据成功' % ((count - 1) * batch_size, count * batch_size))
    print('update 到 fetchmany 获取全量数据所用时间:', (datetime.now() - start_time).seconds) # 16s
conn.close()  # 关闭连接

5、使用服务端游标

#1 逐条处理
with psycopg2.connect(database_connection_string) as conn:
    with conn.cursor(name='name_of_cursor') as cursor:

        cursor.itersize = 20000

        query = "SELECT * FROM ..."
        cursor.execute(query)

        for row in cursor:
         # process row

#2 一次处理多条
while True:
    rows = cursor.fetchmany(100)
    if len(rows) > 0:
        for row in rows:
            # process row
    else:
        break

参考文章

http://initd.org/psycopg/docs…
https://www.cnblogs.com/xiao-…

4. Perl

Perl 是一种功能丰富的计算机编程语言。借取了 C、sed、awk、shell 脚本语言以及很多其他程序语言的特性，其中最重要的特性是它内部集成了正则表达式的功能。它就像 C 一样强大，像 awk、sed 等脚本描述语言一样方便，被 Perl 语言爱好者称之为“一种拥有各种语言功能的梦幻脚本言”、“Unix 中的王牌工具”。

Perl 语言连接 Greenplum 也是通过底层 C 接口 libpq 进行交互，如果我们要使用 Perl 做数据库交互，首先我们需要安装 Perl 的数据库标准模块 Perl DBI，然后再安装相应的 DBD 驱动，这里连接 Greenplum 采用的是 DBD::pg (
https://metacpan.org/pod/DBD::Pg ) 模块。

安装

这里采用较为简单的 RPM 安装方式。

rpm -ivh postgresql-libs-9.2.24-1.el7_5.x86_64.rpm
rpm -ivh perl-DBI-1.627-4.el7.x86_64.rpm
rpm -ivh perl-version-0.99.07-3.el7.x86_64.rpm
rpm -ivh perl-DBD-Pg-2.19.3-4.el7.x86_64.rpm

Perl 连接 Greenplum

连接整体比较简单，DBI 驱动已经将连接方式抽象为易于理解的格式，下面通过几个例子来展示，读者可以关注如何配置连接方式、如何创建连接、如何增删改查等操作。

示例

1、建表

该示例主要演示连接字符串定义，数据库连接打开关闭几在数据库中创建表。

#!/usr/bin/perl

use DBI;
use strict;

my $driver   = "Pg";
my $database = "postgres";
my $dsn = "DBI:$driver:dbname=$database;host=172.16.142.192;port=5432";
my $userid = "gpadmin";
my $password = "gpadmin";
my $dbh = DBI->connect($dsn, $userid, $password, { RaiseError => 1})
                      or die $DBI::errstr;
print "Opened database successfully\n";
my $stmt = qq(CREATE TABLE TC(ID INT, NAME TEXT););
my $rv = $dbh->do($stmt);
if($rv < 0){print $DBI::errstr;} else {print "Table created successfully\n";}
$dbh->disconnect();

执行结果如下：

[root@gp1 ~]# ./create.pl
Opened database successfully
NOTICE:  Table doesn't have'DISTRIBUTED BY'clause -- Using column named'id' as the Greenplum Database data distribution key for this table.
HINT:  The 'DISTRIBUTED BY' clause determines the distribution of data. Make sure column(s) chosen are the optimal data distribution key to minimize skew.
Table created successfully

2、插入数据

#!/usr/bin/perl

use DBI;
use strict;

my $driver   = "Pg";
my $database = "postgres";
my $dsn = "DBI:$driver:dbname=$database;host=172.16.142.192;port=5432";
my $userid = "gpadmin";
my $password = "gpadmin";
my $dbh = DBI->connect($dsn, $userid, $password, { RaiseError => 1})
                      or die $DBI::errstr;
print "Opened database successfully\n";

my $stmt = qq(INSERT INTO TC (ID,NAME)
      VALUES (1, 'Chris'));
my $rv = $dbh->do($stmt) or die $DBI::errstr;

$stmt = qq(INSERT INTO TC (ID,NAME)
      VALUES (2, 'Jenny'));
$rv = $dbh->do($stmt) or die $DBI::errstr;

print "Records created successfully\n";
$dbh->disconnect();

执行结果如下：

[root@gp1 ~]# vim insert.pl
[root@gp1 ~]# chmod +x insert.pl
[root@gp1 ~]# ./insert.pl
Opened database successfully
Records created successfully

3、查询数据

!/usr/bin/perl

use DBI;
use strict;

my $driver   = "Pg";
my $database = "postgres";
my $dsn = "DBI:$driver:dbname=$database;host=172.16.142.192;port=5432";
my $userid = "gpadmin";
my $password = "gpadmin";
my $dbh = DBI->connect($dsn, $userid, $password, { RaiseError => 1})
                      or die $DBI::errstr;
print "Opened database successfully\n";

my $stmt = qq(SELECT id, name from TC;);
my $sth = $dbh->prepare($stmt);
my $rv = $sth->execute() or die $DBI::errstr;
if($rv < 0){print $DBI::errstr;}
while(my @row = $sth->fetchrow_array()) {print "ID =". $row[0] . "\n";
      print "NAME =". $row[1] ."\n";
}
print "Operation done successfully\n";
$dbh->disconnect();

执行结果如下：

[gpadmin@gp1 ~]$ vim select.pl
[gpadmin@gp1 ~]$ chmod +x select.pl
[gpadmin@gp1 ~]$ ./select.pl
Opened database successfully
ID = 2
NAME = Jenny
ID = 1
NAME = Chris
Operation done successfully

有关修改和删除的操作，基本与上面例子一致，只需要将代码中 qq() 中的内容替换为 DELETE 和 UPDATE 语句即可。

正文完
greenplum
发表至： greenplum
2020-06-08
 0
关于greenplum:特性分析-GreenPlum-的并行查询优化策略详解
关于greenplum:基于GreenplumKafka的实时数据处理
关于greenplum:greenplum-常用链接
实战系列Greenplum-版本4到5升级避坑指南