csv

type CatProcess ¶

type CatProcess struct {
}

A process that copies the reader to the writer.

func (p *CatProcess) Run(r Reader, b WriterBuilder, errCh chan<- error)

type CsvToJsonProcess ¶

type CsvToJsonProcess struct {
	BaseObject  string
	StringsOnly bool
}

Given a stream of CSV records, generate a stream of JSON records, one per line. The headers are treated as paths into the resulting JSON object, so a CSV file containing the header foo.bar,baz and the data 1, 2 will be converted into a JSON object like {"foo": {"bar": 1}, "baz": 2}

If column values can be successfully unmarshalled as JSON numbers, booleans, objects or arrays then the value will be encoded as the corresponding JSON object, otherwise it will be encoded as a string. Use --strings to force all column values to be encoded as JSON strings.

func (p *CsvToJsonProcess) Run(reader Reader, encoder *json.Encoder, errCh chan<- error)

type InfluxLineFormatProcess ¶

type InfluxLineFormatProcess struct {
	Measurement string   // the name of the measurement
	Timestamp   string   // the name of the timestamp column
	Format      string   // the format of the timestamp column (for format see documentation of go time.Parse())
	Location    string   // the location in which the timestamp is interpreted (per go time.LoadLocation())
	Tags        []string // the columns to be used as tags
	Values      []string // the columns to be used as values.
}

InfluxLineFormatProcess is a process which converts a CSV file into influx line format.

func (p *InfluxLineFormatProcess) Run(reader Reader, out io.Writer, errCh chan<- error)

type Join ¶

type Join struct {
	LeftKeys   []string // the names of the keys from the left stream
	RightKeys  []string // the names of the keys from the right stream
	Numeric    []string // the names of the keys in the left stream that are numeric keys
	LeftOuter  bool     // perform a left outer join - left rows are copied even if there is no matching right row
	RightOuter bool     // perform a right outer join - right rows are copied even if there is no matching left row
}

A Join can be used to construct a process that will join two streams of CSV records by matching records from each stream on the specified key columns.

func (p *Join) WithRight(r Reader) Process

type JsonToCsvProcess ¶

type JsonToCsvProcess struct {
	BaseObject string
	Header     []string
}

Given a stream of JSON records, generate a stream of csv records.

The columns of the output CSV stream are named by the Header parameter. Each column is interprefed as a path into the JSON input object. If the object at that path is a string, the string is copied into the specified column of the output stream. Otherwise, the json encoding of the object is copied into the specified column of the output stream.

Each object in the input object which is mapped by a CSV column is logically deleted from the input object. If --base-object-key is specified, a JSON encoding of the remaining input object is written into the specified column of the CSV output stream.

func (p *JsonToCsvProcess) Run(decoder *json.Decoder, builder WriterBuilder, errCh chan<- error)

type Pipe ¶

type Pipe interface {
	Builder() WriterBuilder // Builds a Writer for the write end of the pipe
	Reader() Reader         // Returns the Reader for the read end of the pipe
}

Implements a unidirectional channel that can connect a reader process to a writer process.

func NewPipe() Pipe

type Process ¶

type Process interface {
	Run(reader Reader, builder WriterBuilder, errCh chan<- error)
}

A Process is a function that can be run asynchronously that consumes a stream of CSV records provided by reader and writes them into a writer of CSV records as constructed by the specified builder. It signals its successful completion by writing a nil into the specified error channel. An unsuccessful completion is signaled by writing at most one error into the specified error channel.

func NewPipeLine(p []Process) Process

type Reader ¶

type Reader interface {
	// Answers the header.
	Header() []string
	// Answers a channel that iterates over a sequence of Records in the stream. The channel
	// remains open until an error is encountered or until the stream is exhausted.
	C() <-chan Record
	// Answers the error that caused the stream to close, if any.
	Error() error
	// Close the reader and release any resources associated with it.
	Close()
}

Reader provides a reader of CSV streams whose first record is a header describing each field.

func WithCsvReader(r *csv.Reader, c io.Closer) Reader

func WithIoReader(io io.ReadCloser) Reader

func WithIoReaderAndDelimiter(io io.ReadCloser, delimiter rune) Reader

func WithProcess(r Reader, p Process) Reader

type Record ¶

type Record interface {
	// Return the header of the record.
	Header() []string
	// Gets the value of the field specified by the key. Returns the empty string
	// if the field does not exist in the record.
	Get(key string) string
	// Puts the value into the field specified by the key.
	Put(key string, value string)
	// Return the contents of the record as a map. Mutation of the map is not supported.
	AsMap() map[string]string
	// Return the contents of the record as a slice. Mutation of the slice is not supported.
	AsSlice() []string
	// Puts all the matching values from the specified record into the receiving record
	PutAll(r Record)
	// Return true if the receiver and the specified record have the same header.
	SameHeader(r Record) bool
}

Record provides keyed access to the fields of data records where each field of a data record is keyed by the value of the corresponding field in the header record.

func ReadAll(reader Reader) ([]Record, error)

type RecordBuilder ¶

type RecordBuilder func(fields []string) Record

func NewRecordBuilder(header []string) RecordBuilder

type RecordComparator ¶

type RecordComparator func(l, r Record) bool

A RecordComparator is a function that returns true if the left Record is 'less' than the right Record according to some total order.

func AsRecordComparator(comparators []RecordComparator) RecordComparator

type SelectProcess ¶

type SelectProcess struct {
	Keys        []string
	PermuteOnly bool
}

Given a header-prefixed input stream of CSV records select the fields that match the specified key (Key). If PermuteOnly is is specified, all the fields of the input stream are preserved, but the output stream is permuted so that the key fields occupy the left-most fields of the output stream. The remaining fields are preserved in their original order.

func (p *SelectProcess) Run(reader Reader, builder WriterBuilder, errCh chan<- error)

type SortComparator ¶

type SortComparator func(i, j int) bool

A Sort comparator compares two records, identified by i and j, and returns true if the ith record is less than the jth record according to some total order.

type SortKeys ¶

type SortKeys struct {
	Keys     []string // list of columns to use for sorting
	Numeric  []string // list of columns for which a numerical string comparison is used
	Reversed []string // list of columns for which the comparison is reversed
}

Specifies the keys to be used by a CSV sort.

func (p *SortKeys) AsRecordComparator() RecordComparator

func (p *SortKeys) AsRecordComparators() []RecordComparator

func (p *SortKeys) AsSort(data []Record) sort.Interface

func (p *SortKeys) AsSortProcess() *SortProcess

func (p *SortKeys) AsSortable(data []Record) *Sortable

func (p *SortKeys) AsStringProjection() StringProjection

func (p *SortKeys) AsStringSliceComparator() StringSliceComparator

type SortProcess ¶

type SortProcess struct {
	AsSort func(data []Record) sort.Interface
	Keys   []string
}

A process, which given a CSV reader, sorts a stream of Records using the sort specified by the result of the AsSort function. The stream is checked to verify that it has the specified keys.

func (p *SortProcess) Run(reader Reader, builder WriterBuilder, errCh chan<- error)

type Sortable ¶

type Sortable struct {
	Keys        []string
	Data        []Record
	Comparators []SortComparator
}

An adapter that converts a slice of CSV records into an instance of sort.Interface using the specified comparators, in order, to compare records.

func (b *Sortable) AsSortProcess() *SortProcess

func (b *Sortable) Comparator(k string, less StringComparator) SortComparator

func (b *Sortable) Len() int

func (b *Sortable) Less(i, j int) bool

func (b *Sortable) Swap(i, j int)

type StringComparator ¶

type StringComparator func(l, r string) bool

A StringComparator is a function that returns true if the left string is 'less' then the right string according to some total order.

type StringProjection ¶

type StringProjection func(r Record) []string

A StringProjection is a function which produces a slice of strings from a Record.

type StringSliceComparator ¶

type StringSliceComparator func(l, r []string) bool

A StringSliceComparator is a function that returns true if the left slice is 'less' than the right slice according to some total order.

func AsStringSliceComparator(comparators []StringComparator) StringSliceComparator

type SurrogateKeysProcess ¶

type SurrogateKeysProcess struct {
	NaturalKeys  []string
	SurrogateKey string
}

Given a header-prefixed input stream of CSV records and the specification of a natural key (NaturalKeys) generate an augmented, header-prefixed, output stream of CSV records which contains a surrogate key (SurrogateKey) that is derived from the MD5 sum of the natural key.

The surrogate key is constructed by calculating the MD5 hash of the string representation of a CSV record that contains the fields of the natural key of each record.

For example, given the following input CSV stream which has a natural key of Date,Amount,Sequence

Date,Amount,Description,Sequence
2014/12/31,100.0,Payment
2014/12/31,100.0,Payment,1
2014/12/31,85.0,Payment

generate an additional column, KeyMD5, containing a surrogate key that represents the natural key.

Date,Amount,Description,Sequence,KeyMD5
2014/12/31,100.0,Payment,"",bead7c34cf0828efb8a240e262e7afea
2014/12/31,100.0,Payment,1,cc8ab528163236eb1aa4004202ee1935
2014/12/31,85.0,Payment,"",8f4d3a8a05031256a4fa4cf1fadd757b

func (p *SurrogateKeysProcess) Run(reader Reader, builder WriterBuilder, errCh chan<- error)

type UniquifyProcess ¶

type UniquifyProcess struct {
	PartialKeys   []string
	AdditionalKey string
}

Given a header-prefixed input stream of CSV records and the specification of a partial key (PartialKey) formed from one or more of the fields, generate an augmented, header-prefixed, stream of CSV records such that the augmented key of each output record is unique. The field used to ensure uniqueness is specified by the AdditionalKey option.

For example, given the following input with the partial key Date,Amount

Date,Amount,Description
2014/12/31,100.0,Payment
2014/12/31,100.0,Payment
2014/12/31,85.0,Payment

Generate an additional column, Sequence, such that the augmented key Date,Amount,Sequence is unique for all input records.

Date,Amount,Description,Sequence
2014/12/31,100.0,Payment,
2014/12/31,100.0,Payment,1
2014/12/31,85.0,Payment,

func (p *UniquifyProcess) Run(reader Reader, builder WriterBuilder, errCh chan<- error)

type UseTabProcess ¶

type UseTabProcess struct {
	OnRead bool
}

Merely copies records from input to output - delimiting munging is done by the tool.

func (p *UseTabProcess) Run(reader Reader, builder WriterBuilder, errCh chan<- error)

type Writer ¶

type Writer interface {
	Header() []string      // Answer the header of the stream.
	Blank() Record         // Provide a blank record compatible with the stream.
	Write(r Record) error  // Write a single record into the underying stream.
	Error() error          // Return the final error.
	Close(err error) error // Close the writer with the specified error.
}

type WriterBuilder ¶

type WriterBuilder func(header []string) Writer

A constructor for a writer using the specified header. By convention, passing a nil to the Builder returns a Writer which will release any underlying resources held by the builder when Close(error) is called.

func WithCsvWriter(w *encoding.Writer, c io.Closer) WriterBuilder

func WithIoWriter(w io.WriteCloser) WriterBuilder

func WithIoWriterAndDelimiter(w io.WriteCloser, delimiter rune) WriterBuilder