godatapipe

README

Data Pipe

Data pipe copies data from one database table to a table in another database using bulk insert statements or faster methods if available (for example, COPY IN for the Postgres driver).

The destination table is truncated!

Database Support

Database	Driver Name	URI
Postgres	postgres	Example
MS SQL server	mssql	Example

• Also supports any database which has Go drivers (source modification required)

Compiling

• Clone go-datapipe into $GOPATH/src/github.com/literatesnow/go-datapipe
• go get
• go install
• Binary is compiled to $GOPATH/bin/go-datapipe

A Vagrant environment which includes golang can also be used for compiling.

Configuration

Configuration is done by environment variables

Env Var	Description	Default
SRC_DB_DRIVER	Source database driver name
SRC_DB_URI	Source database driver URI
SRC_DB_SELECT_SQL	Select statement to query rows from source database
DST_DB_DRIVER	Destination database driver name
DST_DB_URI	Destination database driver URI
DST_DB_SCHEMA	Destination database schema name
DST_DB_TABLE	Destination database table name (without schema)
MAX_ROW_BUF_SZ	Maximum number of rows to buffer at a time	100
MAX_ROW_TX_COMMIT	Maximum number of rows to process before committing the database transaction	500

Performance

• MAX_ROW_BUF_SZ or MAX_ROW_TX_COMMIT too low could cause slow performance.
• MAX_ROW_BUF_SZ too high could cause memory issues on the machine where this program is running.
• MAX_ROW_TX_COMMIT too high could cause the destination database's transaction logs to fill up.

Example

# Copy all rows where thing >= 100 from SQL server dbo.source_table_name to Postgres master.destination_table_name

export SRC_DB_DRIVER="mssql"
export SRC_DB_URI="server=localhost;port=1433;database=Test;user id=domain\user_name;password=test"
export SRC_DB_SELECT_SQL="SELECT * FROM dbo.source_table_name WHERE thing >= 100"

export DST_DB_DRIVER="postgres"
export DST_DB_URI="user=test password=test dbname=test_db host=localhost"
export DST_DB_SCHEMA="master"
export DST_DB_TABLE="destination_table_name"

export MAX_ROW_BUF_SZ=100
export MAX_ROW_TX_COMMIT=500

./go-datapipe

Documentation

Index

• func Run(ctx context.Context, cfg *Config) (rowCount int, err error)
• type Config
  • func (c *Config) EnvInt(envName string, defaultValue int) (dst int, err error)
  • func (c *Config) EnvStr(envName string) (dst string, err error)
  • func (c *Config) Init() (err error)
• type Insert

Functions

func Run

func Run(ctx context.Context, cfg *Config) (rowCount int, err error)

Types

type Config

type Config struct {
	MaxRowBufSz    int //Maximum number of rows to buffer at a time
	MaxRowTxCommit int //Maximum number of rows to process before committing the database transaction

	SrcConn      *sql.Conn // Source database connection overrides Driver/Uri
	SrcDbDriver  string    //Source database driver name
	SrcDbUri     string    //Source database driver URI
	SrcSelectSql string    //Source database select SQL statement

	DstConn     *sql.Conn // Destination database connection overrides Driver/Uri
	DstDbDriver string    //Destination database driver name
	DstDbUri    string    //Destination database driver URI
	DstSchema   string
	DstTable    string //Destination database table name

	ShowStackTrace bool //Display stack traces on error
}

func (*Config) EnvInt

func (c *Config) EnvInt(envName string, defaultValue int) (dst int, err error)

func (*Config) EnvStr

func (c *Config) EnvStr(envName string) (dst string, err error)

func (*Config) Init

func (c *Config) Init() (err error)

type Insert

type Insert interface {
	Append(ctx context.Context, rows *sql.Rows) (err error)
	Flush(ctx context.Context) (totalRowCount int, err error)
	Close() (err error)
}