From 9ed761edc6e03b6f2bba89676ed2565e90884885 Mon Sep 17 00:00:00 2001 From: dan Date: Mon, 6 Apr 2015 14:51:46 -0700 Subject: [PATCH] re-use the database connection to redshift --- redshift_basic_emitter.go | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/redshift_basic_emitter.go b/redshift_basic_emitter.go index 2026893..6c54dde 100644 --- a/redshift_basic_emitter.go +++ b/redshift_basic_emitter.go @@ -20,7 +20,9 @@ type RedshiftBasicEmtitter struct { Format string Jsonpaths string S3Bucket string + S3Prefix string TableName string + Db *sql.DB } // Emit is invoked when the buffer is full. This method leverages the S3Emitter and @@ -29,20 +31,14 @@ func (e RedshiftBasicEmtitter) Emit(b Buffer, t Transformer) { s3Emitter := S3Emitter{S3Bucket: e.S3Bucket} s3Emitter.Emit(b, t) s3File := s3Emitter.S3FileName(b.FirstSequenceNumber(), b.LastSequenceNumber()) - db, err := sql.Open("postgres", os.Getenv("REDSHIFT_URL")) - if err != nil { - log.Fatal(err) - } - - _, err = db.Exec(e.copyStatement(s3File)) + _, err := e.Db.Exec(e.copyStatement(s3File)) if err != nil { log.Fatal(err) } l4g.Debug("Redshift load completed.") - db.Close() } // Creates the SQL copy statement issued to Redshift cluster. @@ -61,5 +57,6 @@ func (e RedshiftBasicEmtitter) copyStatement(s3File string) string { b.WriteString(fmt.Sprintf("DELIMITER '%v'", e.Delimiter)) } b.WriteString(";") + l4g.Debug(b.String()) return b.String() }