importer

Documentation

Index

• type DataWriter
  • func NewDataWriter(pool *redis.Pool, numWorkers, batchSize int) (*DataWriter, error)
  • func (dw *DataWriter) Close()
  • func (dw *DataWriter) WriteTrip(t *Trip)
• type Importer
  • func NewImporter(connPool *redis.Pool, numWorkers, batchSize int) (*Importer, error)
  • func (i *Importer) Run(resetGraph bool) error
• type Trip
• type TripdataReader
  • func NewTripdataReader(zipUrl string) (*TripdataReader, error)
  • func (r *TripdataReader) Close() error
  • func (r *TripdataReader) Read() (*Trip, error)

Types

type DataWriter

type DataWriter struct {
	// contains filtered or unexported fields
}

DataWriter writes Trips to the RedisGraph. It is optimised for throughput, using concurrent writers, each writing batches of CREATE commands.

func NewDataWriter

func NewDataWriter(pool *redis.Pool, numWorkers, batchSize int) (*DataWriter, error)

func (*DataWriter) Close

func (dw *DataWriter) Close()

Calling Close is important to flush any final batches of Trips.

func (*DataWriter) WriteTrip

func (dw *DataWriter) WriteTrip(t *Trip)

Asynchronously writes a Trip to the graph. Failures will panic.

type Importer

type Importer struct {
	// contains filtered or unexported fields
}

Importer reads trip data from the Citi Bike System Data bucket, and writes Trips to the RedisGraph. It is optimised for throughput.

func NewImporter

func NewImporter(connPool *redis.Pool, numWorkers, batchSize int) (*Importer, error)

func (*Importer) Run

func (i *Importer) Run(resetGraph bool) error

Runs the long-running parallel importer. If resetGraph is true, the graph is deleted before starting.

type Trip

type Trip struct {
	StartTime        time.Time
	StopTime         time.Time
	StartStationId   int
	StartStationName string
	StartStationLat  float64
	StartStationLong float64
	EndStationId     int
	EndStationName   string
	EndStationLat    float64
	EndStationLong   float64
}

A Trip is a parsed row from the Citi Bike System Data records.

type TripdataReader

type TripdataReader struct {
	// contains filtered or unexported fields
}

A TripdataReader downloaded, decompresses, and parses a NYC Bike Trip Data file.

func NewTripdataReader

func NewTripdataReader(zipUrl string) (*TripdataReader, error)

Creates a new TripdataReader. This function downloads the file and decompresses it before returning.

func (*TripdataReader) Close

func (r *TripdataReader) Close() error

func (*TripdataReader) Read

func (r *TripdataReader) Read() (*Trip, error)