Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

processors

package

Go to main page

Versions in this module

v1

v1.0.8

May 23, 2024

v1.0.7

Apr 18, 2024

v1.0.6

Oct 17, 2022

v1.0.5

Aug 18, 2022

v1.0.4

Aug 16, 2022

v1.0.2

Jul 27, 2022

v1.0.1

Jul 22, 2022

v1.0.0

Jul 21, 2022

Changes in this version

+ type BigQueryConfig struct

+ DatasetID string

+ JSONPemPath string

+ ProjectID string

+ type BigQueryReader struct

+ AggregateResults bool

+ ConcurrencyLevel int

+ PageSize int

+ TmpTableName string

+ UnflattenResults bool

+ func NewBigQueryReader(config *BigQueryConfig, query string) *BigQueryReader

+ func NewDynamicBigQueryReader(config *BigQueryConfig, sqlGenerator func(etldata.Payload) (string, error)) *BigQueryReader

+ func (r *BigQueryReader) Concurrency() int

+ func (r *BigQueryReader) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (r *BigQueryReader) ForEachQueryData(d etldata.Payload, killChan chan error, forEach func(d etldata.Payload))

+ func (r *BigQueryReader) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (r *BigQueryReader) String() string

+ type BigQueryWriter struct

+ ConcurrencyLevel int

+ func NewBigQueryWriter(config *BigQueryConfig, tableName string) *BigQueryWriter

+ func NewBigQueryWriterForNewTable(config *BigQueryConfig, tableName string, fields map[string]string) *BigQueryWriter

+ func (w *BigQueryWriter) Concurrency() int

+ func (w *BigQueryWriter) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (w *BigQueryWriter) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (w *BigQueryWriter) String() string

+ func (w *BigQueryWriter) WriteBatch(queuedRows []map[string]interface{}) (err error)

+ type CSVTransformer struct

+ Parameters etlutil.CSVParameters

+ func NewCSVTransformer() *CSVTransformer

+ func (w *CSVTransformer) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (w *CSVTransformer) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (w *CSVTransformer) String() string

+ type CSVWriter struct

+ Parameters etlutil.CSVParameters

+ func NewCSVWriter(w io.Writer) *CSVWriter

+ func (w *CSVWriter) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (w *CSVWriter) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (w *CSVWriter) String() string

+ type FileReader struct

+ func NewFileReader(filename string) *FileReader

+ func (r *FileReader) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (r *FileReader) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (r *FileReader) String() string

+ type FtpWriter struct

+ func NewFtpWriter(host, username, password, path string) *FtpWriter

+ func (f *FtpWriter) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (f *FtpWriter) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (f *FtpWriter) String() string

+ type FuncTransformer struct

+ ConcurrencyLevel int

+ Name string

+ func NewFuncTransformer(transform func(d etldata.Payload) etldata.Payload) *FuncTransformer

+ func (t *FuncTransformer) Concurrency() int

+ func (t *FuncTransformer) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (t *FuncTransformer) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (t *FuncTransformer) String() string

+ type HTTPRequest struct

+ Client *http.Client

+ Request *http.Request

+ func NewHTTPRequest(method, url string, body io.Reader) (*HTTPRequest, error)

+ func (r *HTTPRequest) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (r *HTTPRequest) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (r *HTTPRequest) String() string

+ type IoReader struct

+ BufferSize int

+ Gzipped bool

+ LineByLine bool

+ Reader io.Reader

+ func NewIoReader(reader io.Reader) *IoReader

+ func (r *IoReader) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (r *IoReader) ForEachData(killChan chan error, foo func(d etldata.Payload))

+ func (r *IoReader) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (r *IoReader) String() string

+ type IoReaderWriter struct

+ func NewIoReaderWriter(reader io.Reader, writer io.Writer) *IoReaderWriter

+ func (r *IoReaderWriter) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (r *IoReaderWriter) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (r *IoReaderWriter) String() string

+ type IoWriter struct

+ AddNewline bool

+ Writer io.Writer

+ func NewIoWriter(writer io.Writer) *IoWriter

+ func (w *IoWriter) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (w *IoWriter) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (w *IoWriter) String() string

+ type MySQLWriter struct

+ BatchSize int

+ ConcurrencyLevel int

+ OnDupKeyFields []string

+ OnDupKeyUpdate bool

+ TableName string

+ func NewMySQLWriter(db *sql.DB, tableName string) *MySQLWriter

+ func (s *MySQLWriter) Concurrency() int

+ func (s *MySQLWriter) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (s *MySQLWriter) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (s *MySQLWriter) String() string

+ type Passthrough struct

+ func NewPassthrough() *Passthrough

+ func (r *Passthrough) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (r *Passthrough) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (r *Passthrough) String() string

+ type PostgreSQLWriter struct

+ BatchSize int

+ ConcurrencyLevel int

+ OnDupKeyFields []string

+ OnDupKeyIndex string

+ OnDupKeyUpdate bool

+ TableName string

+ func NewPostgreSQLWriter(db *sql.DB, tableName string) *PostgreSQLWriter

+ func (s *PostgreSQLWriter) Concurrency() int

+ func (s *PostgreSQLWriter) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (s *PostgreSQLWriter) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (s *PostgreSQLWriter) String() string

+ type RedshiftWriter struct

+ BatchSize int

+ Compress bool

+ FileNameWidth int

+ S3IamRole string

+ func NewRedshiftWriter(tx *sql.Tx, config *aws.Config, tableName, bucket, prefix string) *RedshiftWriter

+ func (r *RedshiftWriter) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (r *RedshiftWriter) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (r *RedshiftWriter) String() string

+ type RegexpMatcher struct

+ DebugLog bool

+ func NewRegexpMatcher(pattern string) *RegexpMatcher

+ func (r *RegexpMatcher) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (r *RegexpMatcher) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (r *RegexpMatcher) String() string

+ type S3Reader struct

+ DeleteObjects bool

+ func NewS3ObjectReader(awsID, awsSecret, awsRegion, bucket, object string) *S3Reader

+ func NewS3PrefixReader(awsID, awsSecret, awsRegion, bucket, prefix string) *S3Reader

+ func (r *S3Reader) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (r *S3Reader) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (r *S3Reader) String() string

+ type S3Writer struct

+ Compress bool

+ LineSeparator string

+ func NewS3Writer(awsID, awsSecret, awsRegion, bucket, key string) *S3Writer

+ func (w *S3Writer) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (w *S3Writer) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (w *S3Writer) String() string

+ type SCP struct

+ Destination string

+ Object string

+ Port string

+ func NewSCP(obj string, destination string) *SCP

+ func (s *SCP) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (s *SCP) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (s *SCP) Run(killChan chan error)

+ type SQLExecutor struct

+ func NewDynamicSQLExecutor(dbConn *sql.DB, sqlGenerator func(etldata.Payload) (string, error)) *SQLExecutor

+ func NewSQLExecutor(dbConn *sql.DB, sql string) *SQLExecutor

+ func (s *SQLExecutor) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (s *SQLExecutor) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (s *SQLExecutor) String() string

+ type SQLReader struct

+ BatchSize int

+ ConcurrencyLevel int

+ StructDestination interface{}

+ func NewDynamicSQLReader(dbConn *sql.DB, sqlGenerator func(etldata.Payload) (string, error)) *SQLReader

+ func NewSQLReader(dbConn *sql.DB, sql string) *SQLReader

+ func (s *SQLReader) Concurrency() int

+ func (s *SQLReader) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (s *SQLReader) ForEachQueryData(d etldata.Payload, killChan chan error, forEach func(d etldata.Payload))

+ func (s *SQLReader) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (s *SQLReader) String() string

+ type SQLReaderMySQLWriter struct

+ ConcurrencyLevel int

+ func NewDynamicSQLReaderMySQLWriter(readConn *sql.DB, writeConn *sql.DB, ...) *SQLReaderMySQLWriter

+ func NewSQLReaderMySQLWriter(readConn *sql.DB, writeConn *sql.DB, readQuery, writeTable string) *SQLReaderMySQLWriter

+ func (s *SQLReaderMySQLWriter) Concurrency() int

+ func (s *SQLReaderMySQLWriter) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (s *SQLReaderMySQLWriter) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (s *SQLReaderMySQLWriter) String() string

+ type SQLReaderPostgreSQLWriter struct

+ ConcurrencyLevel int

+ func NewDynamicSQLReaderPostgreSQLWriter(readConn *sql.DB, writeConn *sql.DB, ...) *SQLReaderPostgreSQLWriter

+ func NewSQLReaderPostgreSQLWriter(readConn *sql.DB, writeConn *sql.DB, readQuery, writeTable string) *SQLReaderPostgreSQLWriter

+ func (s *SQLReaderPostgreSQLWriter) Concurrency() int

+ func (s *SQLReaderPostgreSQLWriter) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (s *SQLReaderPostgreSQLWriter) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (s *SQLReaderPostgreSQLWriter) String() string

+ type SQLWriterData struct

+ InsertData interface{}

+ TableName string

+ type SftpReader struct

+ CloseOnFinish bool

+ DeleteObjects bool

+ FileNamesOnly bool

+ Walk bool

+ func NewSftpReader(server string, username string, path string, authMethods ...ssh.AuthMethod) *SftpReader

+ func NewSftpReaderByClient(client *sftp.Client, path string) *SftpReader

+ func (r *SftpReader) CloseClient()

+ func (r *SftpReader) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (r *SftpReader) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (r *SftpReader) String() string

+ type SftpWriter struct

+ CloseOnFinish bool

+ func NewSftpWriter(server string, username string, path string, authMethods ...ssh.AuthMethod) *SftpWriter

+ func NewSftpWriterByFile(file *sftp.File) *SftpWriter

+ func (w *SftpWriter) Finish(outputChan chan etldata.Payload, killChan chan error)

+ func (w *SftpWriter) ProcessData(d etldata.Payload, outputChan chan etldata.Payload, killChan chan error)

+ func (w *SftpWriter) String() string