Complex Pipelines in Go (Part 1): Introduction
Tags: go, postgres, tutorial
Published: 2020-11-11 04:43:06

Just recently I had the opportunity to work on a tool in charge of processing gigabytes of data over the wire, the end goal was to download that data, process the values and finally insert them into persistent storage **in batches**.

This is the first of a series of posts covering all the different pieces involved to achieve the final tool.

### Big Picture

Succinctly this solution will consist of 3 processes:

1. Data Producer Process: reads input data and sends that to a destination for further processing,
1. Data Consumer Process: receives raw data, parses those values using the expected format and sends them to a different process,
1. Persistent Storage Process: receives parsed data and stores that in batches.

This is the classic problem solved using [Pipelines](https://blog.golang.org/pipelines). The biggest difference between that classic post and this new series is how cancellation comes into place when working with multiple goroutines. This means defining rules regarding the expected behavior when anything fails, all of this handled using two great Go packages: [`context`](https://pkg.go.dev/context?tab=doc) and [`errgroup`](https://pkg.go.dev/golang.org/x/sync/errgroup?tab=doc).

For our example we will be using a file part of [IMDB's datasets](https://datasets.imdbws.com/). Those files are gzipped, tab-separated-values (TSV) formatted in the UTF-8 character set. The specific file to use will be [`name.basics.tsv.gz`](https://datasets.imdbws.com/name.basics.tsv.gz) which defines the following fields:

```go
|-------------------|-----------|---------------------------------------------------|
|       Field       | Data Type |                   Description                     |
|-------------------|-----------|---------------------------------------------------|
| nconst            | string    | alphanumeric unique identifier of the name/person |
| primaryName       | string    | name by which the person is most often credited   |
| birthYear         | string    | in YYYY format                                    |
| deathYear         | string    | in YYYY format if applicable, else '\N'           |
| primaryProfession | []string  | the top-3 professions of the person               |
| knownForTitles    | []string  | titles the person is known for                    |
|-------------------|-----------|---------------------------------------------------|
```

### Data Producer Process: Input Data Format

Because of the location (http resource) and the file data format (gzip) of this input file, our _Data Producer Process_ will request the file using [`net/http`](https://pkg.go.dev/net/http?tab=doc), uncompres the received values using [`compress/gzip`](https://pkg.go.dev/compress/gzip?tab=doc) and send them to the _Data Consumer Process_ as raw `[]byte`.

### Data Consumer Process: Input Data Format

Those raw `[]byte` values will be read into TSV records using [`encoding/csv`](https://pkg.go.dev/encoding/csv?tab=doc) and from there they will be converted into values of a new struct type `Name` that our next step in our pipeline can understand.

### Persistent Storage Process: Input Data Format

The following will be used as the type containing the values to be eventually persisted in a database:

```go
type Name struct {
	NConst             string
	PrimaryName        string
	BirthYear          string
	DeathYear          string
	PrimaryProfessions []string
	KnownForTitles     []string
}
```

We will be using PostgreSQL as the relational database for this, and specifically [`github.com/jackc/pgx`](https://pkg.go.dev/github.com/jackc/pgx?tab=doc) will be imported for storing the values in batches.

### What's next?

The next blog pos will cover the implementation of the PostgreSQL Batcher, as we progress in the series we will continuously connect all the pieces together to eventually complete our final tool.