助力Greenplum数据库开发之接口篇（上）——Golang，C、Python和Perl

74 阅读 0 评论 49 点赞

我是靠谱客的博主平淡红牛，这篇文章主要介绍助力Greenplum数据库开发之接口篇（上）——Golang，C、Python和Perl，现在分享给大家，希望可以做个参考。

了解更多Greenplum技术干货，欢迎访问Greenplum中文社区网站

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 连接数据库查询的代码，进行简单分析。

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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”。

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

直接查询
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")
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
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-index/ ）中，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("nn");

/* 打印数据 */
    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;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
/*
 * 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("nn");

    /* 打印数据 */
    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/docs/current/libpq-example.html

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 版本。

安装

复制代码

1
2
pip install PyGreSQL

示例

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
#!/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.html

Psycopg2

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

安装

复制代码

1
2
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() # 关闭连接

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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、批量插入，查询

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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()  # 关闭连接

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
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()  # 关闭连接

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#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/index.html
https://www.cnblogs.com/xiao-apple36/p/10362367.html

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 安装方式。

复制代码

1
2
3
4
5
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、建表

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

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#!/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 successfullyn";
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 successfullyn";
}
$dbh->disconnect();

执行结果如下：

复制代码

1
2
3
4
5
6
[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、插入数据

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
#!/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 successfullyn";

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 successfullyn";
$dbh->disconnect();

执行结果如下：

复制代码

1
2
3
4
5
6
[root@gp1 ~]# vim insert.pl
[root@gp1 ~]# chmod +x insert.pl
[root@gp1 ~]# ./insert.pl
Opened database successfully
Records created successfully

3、查询数据

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
!/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 successfullyn";

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 successfullyn";
$dbh->disconnect();

执行结果如下：

复制代码

1
2
3
4
5
6
7
8
9
10
[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数据库开发之接口篇（上）——Golang，C、Python和Perl的全部内容，更多相关助力Greenplum数据库开发之接口篇（上）——Golang内容请搜索靠谱客的其他文章。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。

本文分类：数据库
浏览次数：74 次浏览
发布日期：2023-12-28 20:35:21
本文链接：https://www.kaopuke.com/article/k-p-k_13_u_23_ocfy_13__7__18_4.html

助力Greenplum数据库开发之接口篇（上）——Golang，C、Python和Perl

1. Golang

支持列表

安装方法

使用方法

2. C语言

示例

编译

参考文章

3. Python

PyGreSQL

安装

示例

参考文章

Psycopg2

安装

示例

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

2、批量插入，查询

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

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

5、使用服务端游标

参考文章

4. Perl

安装

Perl 连接 Greenplum

示例

1、建表

2、插入数据

3、查询数据

最后

评论列表共有 0 条评论

发表评论取消回复

助力Greenplum数据库开发之接口篇（上）——Golang，C、Python和Perl

1. Golang

支持列表

安装方法

使用方法

2. C语言

示例

编译

参考文章

3. Python

PyGreSQL

安装

示例

参考文章

Psycopg2

安装

示例

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

2、批量插入，查询

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

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

5、使用服务端游标

参考文章

4. Perl

安装

Perl 连接 Greenplum

示例

1、建表

2、插入数据

3、查询数据

最后

相关文章

评论列表共有 0 条评论

发表评论 取消回复

微信扫一扫：分享

发表评论取消回复