pg_bulkload批量载入工具初探



安装：

#tar-zxvfpg_bulkload-3.1.6.tar.gz
#cdpg_bulkload-3.1.6
#./home/postgres/.bash_profile
#make
#makeinstall

引入扩展：

$psql
psql(9.3.4)
Type"help"forhelp.
postgres=#createdatabasetest;
CREATEDATABASE
postgres=#\ctest
YouareNowconnectedtodatabase"test"asuser"postgres".
test=#createextensionpg_bulkload;
CREATEEXTENSION
test=#\dx
listofinstalledextensions
Name|Version|Schema|Description
-------------+---------+------------+-----------------------------------------------------------------
pg_bulkload|1.0|public|pg_bulkloadisahighspeeddataloadingutilityforPostgresql
plpgsql|1.0|pg_catalog|PL/pgsqlprocedurallanguage
(2rows)
test=#\df
listoffunctions
Schema|Name|Resultdatatype|Argum
entdatatypes|Type
--------+-------------+------------------+------------------------------------------------------------------------------------------------------------
---------------------------------------------------------------------------------------------------------------------+--------
public|pg_bulkload|record|optionstext[],OUTskipbigint,OUTcountbigint,OUTparse_errorsbigint,OUTduplicate_newbigint,OUTd
uplicate_oldbigint,OUTsystem_timedoubleprecision,OUTuser_timedoubleprecision,OUTdurationdoubleprecision|normal
(1row)

创造测试数据：

test=#createtablet1(idint,nametext);
CREATETABLE
test=#insertintot1selectgenerate_series(1,50000000),'HighGo';
INSERT050000000
test=#\d+
listofrelations
Schema|Name|Type|Owner|Size|Description
--------+------+-------+----------+---------+-------------
public|t1|table|postgres|2112MB|
(1row)
test=#copyt1to'/opt/pg93/data/t1.csv'with(format'csv');
copY50000000

$du-sh/opt/pg93/data/t1.csv
754M/opt/pg93/data/t1.csv

copy方式载入数据：

(with logged):

test=#truncatet1;
TruncATETABLE
test=#\timing
Timingison.
test=#
test=#copyt1from'/opt/pg93/data/t1.csv'with(format'csv');
copY50000000
Time:139038.099ms

(without logged):

test=#truncatet1;
TruncATETABLE

修改t1表为unlogged
test=#updatepg_classsetrelpersistence='u'whererelname='t1';
UPDATE1

test=#copyt1from'/opt/pg93/data/t1.csv'with(format'csv');
copY50000000
Time:110796.480ms

pg_blukload载入数据：

(without logged):

$pg_bulkload-I/Opt/pg93/data/t1.csv-Ot1-lt1_bulkload.log-o"TYPE=CSV"-o"WRITER=ParaLLEL"-dtest
NOTICE:BULKLOADSTART
NOTICE:BULKLOADEND
0Rowsskipped.
50000000Rowssuccessfullyloaded.
0Rowsnotloadedduetoparseerrors.
0Rowsnotloadedduetoduplicateerrors.
0Rowsreplacedwithnewrows.
Runbeganon2014-06-1605:53:45.025377+08
Runendedon2014-06-1605:55:00.625057+08
cpu2.80s/33.00usecelapsed75.60sec

$pg_bulkload-I/Opt/pg93/data/t1.csv-Ot1-lt1_bulkload.log-o"TYPE=CSV"-o"WRITER=DIRECT"-dtest
NOTICE:BULKLOADSTART
NOTICE:BULKLOADEND
0Rowsskipped.
50000000Rowssuccessfullyloaded.
0Rowsnotloadedduetoparseerrors.
0Rowsnotloadedduetoduplicateerrors.
0Rowsreplacedwithnewrows.
Runbeganon2014-06-1606:05:40.267198+08
Runendedon2014-06-1606:07:05.08921+08
cpu6.88s/34.25usecelapsed84.82sec

(with logged):

test=#truncatet1;
TruncATETABLE
修改t1表为logged
test=#updatepg_classsetrelpersistence='p'whererelname='t1';
UPDATE1

$pg_bulkload-I/Opt/pg93/data/t1.csv-Ot1-lt1_bulkload.log-o"TYPE=CSV"-o"WRITER=ParaLLEL"-dtest
NOTICE:BULKLOADSTART
NOTICE:BULKLOADEND
0Rowsskipped.
50000000Rowssuccessfullyloaded.
0Rowsnotloadedduetoparseerrors.
0Rowsnotloadedduetoduplicateerrors.
0Rowsreplacedwithnewrows.
Runbeganon2014-06-1605:57:05.620751+08
Runendedon2014-06-1605:58:20.458029+08
cpu2.80s/33.02usecelapsed74.84sec


$pg_bulkload-I/Opt/pg93/data/t1.csv-Ot1-lt1_bulkload.log-o"TYPE=CSV"-o"WRITER=DIRECT"-dtest
NOTICE:BULKLOADSTART
NOTICE:BULKLOADEND
0Rowsskipped.
50000000Rowssuccessfullyloaded.
0Rowsnotloadedduetoparseerrors.
0Rowsnotloadedduetoduplicateerrors.
0Rowsreplacedwithnewrows.
Runbeganon2014-06-1606:02:10.33344+08
Runendedon2014-06-1606:03:36.986382+08
cpu7.15s/34.93usecelapsed86.65sec

pg_bulkload执行过程中查看进程和连接：
（WRITER=ParaLLEL）

$ps-ef|greppost
postgres2404422690005:53pts/200:00:00pg_bulkload-I/Opt/pg93/data/t1.csv-Ot1-lt1_bulkload.log-oTYPE=CSV-oWRITER=ParaLLEL-dtest
postgres2404522364805:53?00:00:04postgres:postgrestest[local]SELECT
postgres2404622362005:53?00:00:02postgres:postgrestest[local]SELECT

postgres=#selectdatname,application_name,queryfrompg_stat_activity;
datname|application_name|query


----------+------------------+------------------------------------------------------------------------------------------------------------------------
-----------------------------------------------------------------------------------------------------------
postgres|psql|selectdatname,queryfrompg_stat_activity;
test|pg_bulkload|SELECT*FROMpg_bulkload($1)
test||SELECT*FROMpg_bulkload(ARRAY['TYPE=TUPLE','INPUT='||$1,'WRITER=DIRECT','OUTPUT='||$2,'ON_DUPLICATE_KEEP='||$3,'DUPLICATE_ERRORS='||$4,'DUPLICATE_BADFILE='||$5,'LOGFILE='||$6,'VERBOSE='||$7,'TruncATE='||$8])
(3rows)

（WRITER=ParaLLEL）

$ps-ef|greppost
postgres2412422690006:02pts/200:00:00pg_bulkload-I/Opt/pg93/data/t1.csv-Ot1-lt1_bulkload.log-oTYPE=CSV-oWRITER=DIRECT-dtest
postgres2412522364906:02?00:00:13postgres:postgrestest[local]SELECT

test=#selectdatname,queryfrompg_stat_activity;
datname|application_name|query
---------+------------------+--------------------------------------------------------------
test|pg_bulkload|SELECT*FROMpg_bulkload($1)
test|psql|selectdatname,queryfrompg_stat_activity;
(2rows)