TBase应用接入指南(上篇)

PostgreSQL中文社区 2019-12-10

1228

版本号－－V2.17

更新日期：2019-11-13

1、JAVA开发

1.1、创建数据

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;




public class createtable {
public static void main( String args[] )
     {
Connection c = null;
Statement stmt = null;
try {
Class.forName("org.postgresql.Driver");
c = DriverManager.getConnection("jdbc:postgresql://127.0.0.1:15432/postgres?currentSchema=public&binaryTransfer=false","tbase", "tbase");
System.out.println("Opened database successfully");
         stmt = c.createStatement();
String sql = "create table tbase(id int,nickname text) distribute by shard(id) to group  default_group" ;
         stmt.executeUpdate(sql);
         stmt.close();
c.close();
       } catch ( Exception e ) {
System.err.println( e.getClass().getName()+": "+ e.getMessage() );
System.exit(0);
       }
System.out.println("Table created successfully");
     }
}

说明：

n 这里连接的节点为任意CN主节点，后面所有操作，没特别说明，都是连接到CN主节点进行操作。

1.2、使用普通协议插入数据

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;


public class insert {
public static void main(String args[]) {
Connection c = null;
Statement stmt = null;
try {
Class.forName("org.postgresql.Driver");
c = DriverManager.getConnection("jdbc:postgresql://127.0.0.1:15432/postgres?currentSchema=public&binaryTransfer=false","tbase", "tbase");
c.setAutoCommit(false);
System.out.println("Opened database successfully");


         stmt = c.createStatement();
String sql = "INSERT INTO tbase (id,nickname) "
               + "VALUES (1,'tbase');";
         stmt.executeUpdate(sql);


         sql = "INSERT INTO tbase (id,nickname) "
               + "VALUES (2, 'pgxz' ),(3,'pgxc');";
         stmt.executeUpdate(sql);
         stmt.close();
c.commit();
c.close();
      } catch (Exception e) {
System.err.println( e.getClass().getName()+": "+ e.getMessage() );
System.exit(0);
      }
System.out.println("Records created successfully");
   }
}

1.3、使用扩展协议插入数据

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.*;
import java.util.Random;


public class insert_prepared {
public static void main(String args[]) {
Connection c = null;
PreparedStatement stmt;
try {
Class.forName("org.postgresql.Driver");
c = DriverManager.getConnection("jdbc:postgresql://127.0.0.1:15432/postgres?currentSchema=public&binaryTransfer=false","tbase", "tbase");
c.setAutoCommit(false);
System.out.println("Opened database successfully");
//插入数据
String sql = "INSERT INTO tbase (id,nickname) VALUES (?,?)";         
         stmt = c.prepareStatement(sql);
         stmt.setInt(1, 9999);
         stmt.setString(2, "tbase_prepared");
         stmt.executeUpdate();


//插入更新
         sql = "INSERT INTO tbase (id,nickname) VALUES (?,?) ON CONFLICT(id) DO UPDATE SET nickname=?";
         stmt = c.prepareStatement(sql);
         stmt.setInt(1, 9999);
         stmt.setString(2, "tbase_prepared");
         stmt.setString(3, "tbase_prepared_update");
         stmt.executeUpdate();


         stmt.close();
c.commit();
c.close();
      } catch (Exception e) {
System.err.println( e.getClass().getName()+": "+ e.getMessage() );
System.exit(0);
      }
System.out.println("Records created successfully");
   }
}

1.4、copy from 加载文件到表

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import org.postgresql.copy.CopyManager;  
import org.postgresql.core.BaseConnection;  
import java.io.*;


public class copyfrom {
public static void main( String args[] )
     {
Connection c = null;
Statement stmt = null;
FileInputStream fs = null;
try {
Class.forName("org.postgresql.Driver");
c = DriverManager.getConnection("jdbc:postgresql://127.0.0.1:15432/postgres?currentSchema=public&binaryTransfer=false","tbase", "tbase");
System.out.println("Opened database successfully");
CopyManager cm = new CopyManager((BaseConnection) c);
         fs = new FileInputStream("/data/tbase/tbase.csv");
String sql = "COPY tbase FROM STDIN DELIMITER AS ','";
         cm.copyIn(sql, fs);
c.close();
         fs.close();
       } catch ( Exception e ) {
System.err.println( e.getClass().getName()+": "+ e.getMessage() );
System.exit(0);
       }
System.out.println("Copy data successfully");
     }
}

1.5、copy to 导出数据到文件

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import org.postgresql.copy.CopyManager;  
import org.postgresql.core.BaseConnection;  
import java.io.*;


public class copyto {
public static void main( String args[] )
     {
Connection c = null;
Statement stmt = null;
FileOutputStream fs = null;
try {
Class.forName("org.postgresql.Driver");
c = DriverManager.getConnection("jdbc:postgresql://127.0.0.1:15432/postgres?currentSchema=public&binaryTransfer=false","tbase", "tbase");
System.out.println("Opened database successfully");
CopyManager cm = new CopyManager((BaseConnection) c);
         fs = new FileOutputStream("/data/tbase/tbase.csv");
String sql = "COPY tbase TO STDOUT DELIMITER AS ','";
         cm.copyOut(sql, fs);
c.close();
         fs.close();
       } catch ( Exception e ) {
System.err.println( e.getClass().getName()+": "+ e.getMessage() );
System.exit(0);
       }
System.out.println("Copy data successfully");
     }
}

1.6、jdbc包下载地址

请点击文章底部“阅读原文”进行下载

2、C程序开发

2.1、连接数据库

#include <stdio.h>
#include <stdlib.h>
#include "libpq-fe.h"
int
main(int argc, char **argv){
const char *conninfo;
    PGconn     *conn;      
if (argc > 1){
        conninfo = argv[1];
    }else{
        conninfo = "dbname = postgres";  
    }            
    conn = PQconnectdb(conninfo);
if (PQstatus(conn) != CONNECTION_OK){
fprintf(stderr, "连接数据库失败: %s",PQerrorMessage(conn));              
    }else{
printf("连接数据库成功！\n");
    }
    PQfinish(conn);
return 0;
}


编译


gcc -c -I usr/local/install/tbase_pgxz/include/ conn.c
gcc -o conn conn.o -L usr/local/install/tbase_pgxz/lib/ -lpq


运行


./conn "host=172.16.0.3 dbname=postgres port=11000"
连接数据库成功！
./conn "host=172.16.0.3 dbname=postgres port=15432 user=tbase"
连接数据库成功！

2.2、建立数据表

#include <stdio.h>
#include <stdlib.h>
#include "libpq-fe.h"
int
main(int argc, char **argv){
const char *conninfo;
    PGconn     *conn;      
    PGresult   *res;
const char *sql = "create table tbase(id int,nickname text) distribute by shard(id) to group  default_group";
if (argc > 1){
        conninfo = argv[1];
    }else{
        conninfo = "dbname = postgres";           
    }        
    conn = PQconnectdb(conninfo);
if (PQstatus(conn) != CONNECTION_OK){
fprintf(stderr, "连接数据库失败: %s",PQerrorMessage(conn));              
    }else{
printf("连接数据库成功！\n");
    }
    res = PQexec(conn,sql);
if(PQresultStatus(res) != PGRES_COMMAND_OK){
fprintf(stderr, "建立数据表失败: %s",PQresultErrorMessage(res)); 
    }else{
printf("建立数据表成功！\n");
    }
    PQclear(res);
    PQfinish(conn);
return 0;
}


编译


gcc -c -I usr/local/install/tbase_pgxz/include/ createtable.c
gcc -o createtable createtable.o -L usr/local/install/tbase_pgxz/lib/ -lpq


运行


./createtable "port=11000 dbname=postgres"
连接数据库成功！
建立数据表成功！

2.3、插入数据

#include <stdio.h>
#include <stdlib.h>
#include "libpq-fe.h"
int
main(int argc, char **argv){
const char *conninfo;
    PGconn     *conn;      
    PGresult   *res;
const char *sql = "INSERT INTO tbase (id,nickname) values(1,'tbase'),(2,'pgxz')";
if (argc > 1){
        conninfo = argv[1];
    }else{
        conninfo = "dbname = postgres";           
    }        
    conn = PQconnectdb(conninfo);
if (PQstatus(conn) != CONNECTION_OK){
fprintf(stderr, "连接数据库失败: %s",PQerrorMessage(conn));              
    }else{
printf("连接数据库成功！\n");
    }
    res = PQexec(conn,sql);
if(PQresultStatus(res) != PGRES_COMMAND_OK){
fprintf(stderr, "插入数据失败: %s",PQresultErrorMessage(res)); 
    }else{
printf("插入数据成功！\n");
    }
    PQclear(res);
    PQfinish(conn);
return 0;
}


编译


gcc -c -I usr/local/install/tbase_pgxz/include/ insert.c
gcc -o insert insert.o -L usr/local/install/tbase_pgxz/lib/ -lpq


运行
./insert "dbname=postgres port=15432"
连接数据库成功！
插入数据成功！

2.4、查询数据

#include <stdio.h>
#include <stdlib.h>
#include "libpq-fe.h"   
int
main(int argc, char **argv){
    const char *conninfo;
    PGconn     *conn;      
    PGresult   *res;
    const char *sql = "select * from tbase";
    if (argc > 1){
        conninfo = argv[1];
    }else{
        conninfo = "dbname = postgres";           
    }
    conn = PQconnectdb(conninfo);    
    if (PQstatus(conn) != CONNECTION_OK){
        fprintf(stderr, "连接数据库失败: %s",PQerrorMessage(conn));              
    }else{    
        printf("连接数据库成功！\n");
    }                                
    res = PQexec(conn,sql);
    if(PQresultStatus(res) != PGRES_TUPLES_OK){
        fprintf(stderr, "插入数据失败: %s",PQresultErrorMessage(res)); 
    }else{
        printf("查询数据成功！\n");    
        int rownum = PQntuples(res) ;
        int colnum = PQnfields(res);
        for(int j = 0;j< colnum; ++j){
            printf("%s\t",PQfname(res,j));
        }
        printf("\n");
        for(int i = 0;i< rownum; ++i){
            for(int j = 0;j< colnum; ++j){
                printf("%s\t",PQgetvalue(res,i,j));
            }
            printf("\n");
        }
    }
    PQclear(res);
    PQfinish(conn);
    return 0;
}
 
编译
 
gcc -std=c99 -c -I usr/local/install/tbase_pgxz/include/ select.c
gcc -o select select.o -L usr/local/install/tbase_pgxz/lib/ -lpq
 
运行
 
./select "dbname=postgres port=15432"
连接数据库成功！
查询数据成功！
id      nickname
1       tbase
2       pgxz

‍

2.5、流数据COPY入表

#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include "libpq-fe.h"
int
main(int argc, char **argv){
const char *conninfo;
    PGconn     *conn;      
    PGresult   *res;
const char *buffer = "1,tbase\n2,pgxz\n3,Tbase牛";
if (argc > 1){
        conninfo = argv[1];
    }else{
        conninfo = "dbname = postgres";           
    }
    conn = PQconnectdb(conninfo);
if (PQstatus(conn) != CONNECTION_OK){
fprintf(stderr, "连接数据库失败: %s",PQerrorMessage(conn));              
    }else{
printf("连接数据库成功！\n");
    }
    res=PQexec(conn,"COPY tbase FROM STDIN DELIMITER ',';");
if(PQresultStatus(res) != PGRES_COPY_IN){
fprintf(stderr, "copy数据出错1: %s",PQresultErrorMessage(res));
    }else{
int len = strlen(buffer);
if(PQputCopyData(conn,buffer,len) == 1){
if(PQputCopyEnd(conn,NULL) == 1){
                res = PQgetResult(conn);
if(PQresultStatus(res) == PGRES_COMMAND_OK){
printf("copy数据成功！\n");         
                }else{
fprintf(stderr, "copy数据出错2: %s",PQerrorMessage(conn));    
                }
             }else{
fprintf(stderr, "copy数据出错3: %s",PQerrorMessage(conn));   
             }
        }else{
fprintf(stderr, "copy数据出错4: %s",PQerrorMessage(conn));              
        }
    }
    PQclear(res);
    PQfinish(conn);
return 0;
}


编译
gcc -c -I usr/local/install/tbase_pgxz/include/ copy.c
gcc -o copy copy.o -L usr/local/install/tbase_pgxz/lib/ -lpq


执行
./copy "dbname=postgres port=15432"
连接数据库成功！
copy数据成功！

3、shell脚本开发

#!/bin/sh


if [ $# -ne 0 ]
then
echo "usage: $0 exec_sql"
exit 1
fi


exec_sql=$1


masters=`psql -h 172.16.0.29 -d postgres -p 15432 -t -c "select string_agg(node_host, ' ') from (select * from pgxc_node where node_type = 'D' order by node_name) t"`
port_list=`psql -h 172.16.0.29 -d postgres -p 15432 -t -c "select string_agg(node_port::text, ' ') from (select * from pgxc_node where node_type = 'D' order by node_name) t"`
node_cnt=`psql -h 172.16.0.29 -d postgres -p 15432 -t -c "select count(*) from pgxc_node where node_type = 'D'"`
masters=($masters)
ports=($port_list)


echo $node_cnt


flag=0


for((i=0;i<$node_cnt;i++));
do
    seq=$(($i+1))
    master=${masters[$i]}
    port=${ports[$i]}
echo $master
echo $port


    psql -h $master -p $port  postgres -c "$exec_sql"
done

4、python程序开发

4.1、安装psycopg2模块

[root@VM_0_29_centos ~]# yum install python-psycopg2

4.2、连接数据库

#coding=utf-8
#!/usr/bin/python
import psycopg2
try:
    conn = psycopg2.connect(database="postgres", user="tbase", password="", host="172.16.0.29", port="15432")
print "连接数据库成功"
    conn.close()
except psycopg2.Error,msg:
print "连接数据库出错，错误详细信息：%s" %(msg.args[0])


运行


[tbase@VM_0_29_centos python]$ python conn.py 
连接数据库成功

4.3、创建数据表

#coding=utf-8
#!/usr/bin/python
import psycopg2
try:
    conn = psycopg2.connect(database="postgres", user="tbase", password="", host="172.16.0.29", port="15432")
print "连接数据库成功"
    cur = conn.cursor()
    sql = """
          create table tbase 
          (
              id int,
              nickname varchar(100)
          )distribute by shard(id) to group default_group
          """
    cur.execute(sql)
    conn.commit()
print "建立数据表成功"
    conn.close()
except psycopg2.Error,msg:
print "TBase Error %s" %(msg.args[0])


运行


[tbase@VM_0_29_centos python]$ python createtable.py 
连接数据库成功
建立数据表成功

4.4、插入数据

#coding=utf-8
#!/usr/bin/python
import psycopg2
try:
    conn = psycopg2.connect(database="postgres", user="tbase", password="", host="172.16.0.29", port="15432")
print "连接数据库成功"
    cur = conn.cursor()
    sql = "insert into tbase values(1,'tbase'),(2,'tbase');"
    cur.execute(sql)
    sql = "insert into tbase values(%s,%s)"
    cur.execute(sql,(3,'pg'))
    conn.commit()
print "插入数据成功"
    conn.close()
except psycopg2.Error,msg:
print "操作数据库出库 %s" %(msg.args[0])


运行


[tbase@VM_0_29_centos python]$ python insert.py 
连接数据库成功
插入数据成功

4.5、查询数据

#coding=utf-8
#!/usr/bin/python
import psycopg2
try:
    conn = psycopg2.connect(database="postgres", user="tbase", password="", host="172.16.0.29", port="15432")
    print "连接数据库成功"
    cur = conn.cursor()
    sql = "select * from tbase"
    cur.execute(sql)
    rows = cur.fetchall()
    for row in rows:
        print "ID = ", row[0]
        print "NICKNAME = ", row[1],"\n"
    conn.close()
except psycopg2.Error,msg:
    print "操作数据库出库 %s" %(msg.args[0])


运行


[tbase@VM_0_29_centos python]$ python select.py 
连接数据库成功
ID =  1
NICKNAME =  tbase 


ID =  2
NICKNAME =  pgxz 


ID =  3
NICKNAME =  pg

4.6、copy from 加载文件到表

#coding=utf-8
#!/usr/bin/python
import psycopg2
try:
    conn = psycopg2.connect(database="postgres", user="tbase", password="", host="172.16.0.29", port="15432")
print "连接数据库成功"
    cur = conn.cursor()
    filename = "/data/tbase/tbase.txt"
    cols = ('id','nickname')
    tablename="public.tbase"
    cur.copy_from(file=open(filename),table=tablename,columns=cols,sep=',')
    conn.commit()
print "导入数据成功"
    conn.close()
except psycopg2.Error,msg:
print "操作数据库出库 %s" %(msg.args[0])


执行


[tbase@VM_0_29_centos python]$ python copy_from.py 
连接数据库成功
导入数据成功

tbase tencentdb

最后修改时间：2019-12-10 10:47:46

文章转载自PostgreSQL中文社区，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。