Building a data pipeline with Ruby on Rails
Tags: ruby, rails, database, architecture
Published: 2020-11-16 15:12:35

If you need to bring a lot of data into your app, a data pipeline can help. This article describes how to build a simple data pipeline with Ruby on Rails, Sidekiq, Redis, and Postgres.

What's a data pipeline?
-----------------------

A data pipeline takes information from one place and puts it somewhere else. It likely transforms that data along the way. In enterprise circles, this is called [ETL](https://en.wikipedia.org/wiki/Extract,_transform,_load), or extract-transform-load.

Generally, data pipelines:

1.  Ingest a lot of data at once
2.  Ingest new or updated data on a schedule

Splitting data into small jobs
------------------------------

A reliable pipeline starts big and reduces data into small slices that can be independently processed. Bigger jobs "fan out" into small ones. The smallest jobs, handling one record each, do most of the work.

Breaking work into smaller jobs helps handle failure. If the pipeline ingests 1,000,000 records, chances are that a few of them are invalid. If those invalid records are processing in their own jobs, then they can fail without impacting other records.

[Lobby Focus](https://www.lobbyfocus.com/), a project of mine, ingests [United States lobbying data from the U.S. Senate](https://www.senate.gov/legislative/Public_Disclosure/database_download.htm). There are about 1,000,000 lobbying records from 2010-2020. Processing them all in a single job would be risky, as the entire job would fail if even one record triggered an error. Instead, Lobby Focus breaks down data into small slices and ends up processing each record in its own job. To bring in 1,000,000 lobbying records, it takes 1,001,041 jobs.

Pipelines can handle a ton of jobs with ease. Splitting up work into many small jobs offers advantages:

1.  Most jobs deal with only one record. If a job fails, it's easy to pin down the problematic record, fix it, and retry it. The rest of the pipeline continues uninterrupted.
2.  The pipeline can ingest data in parallel, with many small jobs running at once.
3.  You can track progress by looking at how many jobs succeeded and how many remain.

A Ruby on Rails pipeline stack
------------------------------

This stack, used in the example below, makes for a solid data pipeline:

-   **[Ruby on Rails](https://rubyonrails.org/)** interfaces with all the pieces below and has a strong ecosystem of libraries.
-   **[Sidekiq](https://github.com/mperham/sidekiq)** manages data pipeline jobs, providing visibility into progress and errors.
-   **[Redis](https://redis.io/)** backs Sidekiq and stores the job queue.
-   **[Postgres](https://www.postgresql.org/)** stores intermediate and final data.
-   **[Amazon S3](https://aws.amazon.com/s3/)** serves as a staging area for source data.

Managed platforms like [Heroku](http://heroku.com/) or [Cloud 66](https://www.cloud66.com/) are good hosting options for this stack.

Real-world example: U.S. Senate lobbying data
---------------------------------------------

[Lobby Focus](https://lobbyfocus.com) ingests lobbying data from the U.S. Senate and makes it accessible to users. Here's a peek into its data pipeline, which breaks data into small jobs and uses the stack above.

### Step 1: Scrape

**1 Sidekiq job** deals with **1,000,000 records**

**Parameter:** None

-   Gathers a list of ZIP files from the U.S. Senate, using [HTTParty](https://github.com/jnunemaker/httparty) to retrieve the webpage and [Nokogiri](https://github.com/sparklemotion/nokogiri) to parse it
-   Launches a download job for each ZIP file. To ingest lobbying data from 2010 to 2020, it will launch 40 **download** jobs for 40 ZIP's.

### Step 2: Download

**40 Sidekiq jobs** deal with **25,000 records** **each**

**Parameter:** ZIP file URL

-   Checks the [last-modified](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Last-Modified) header to see if the current version was already processed. If so, aborts the job.
-   Downloads the ZIP and unzips its contents into Amazon S3, using [rubyzip](https://github.com/rubyzip/rubyzip) and [aws-s3-sdk](https://github.com/aws/aws-sdk-ruby).
-   Launches a **process** job for each XML file in the ZIP.

### Step 3: Process

**1,000 Sidekiq jobs** deal with **1,000 records each**

**Parameter:** XML file URL (S3)

-   Parses the XML with [Nokogiri](https://nokogiri.org/tutorials/parsing_an_html_xml_document.html)
-   Adds the content of each record in the XML file to Postgres, as a raw XML string.
-   Launches a **load** job for each record in the XML file, passing the Postgres record ID.

Postgres can bear a high volume of record inserts. The pipeline uses it as an intermediate data store. The next job, **load**, receives the Postgres record ID and accesses it to get the XML string. Using Postgres to pass the XML avoids passing giant parameters between jobs. Sidekiq [recommends small job parameters](https://github.com/mperham/sidekiq/wiki/Best-Practices) to keep the queue lean and preserve Redis memory.

### Step 4: Load

**1,000,000 Sidekiq jobs** deal with **1 record each**

**Parameter:** Postgres record ID

-   Retrieves the XML string from Postgres and parses it with [Nokogiri](https://nokogiri.org/tutorials/parsing_an_html_xml_document.html)
-   Validates data and aborts the job with an error if there is an issue
-   Upserts final data record into Postgres

Staying resilient
-----------------

Breaking out data loads into small jobs is key to a resilient pipeline. Here are some more tips:

### Make jobs idempotent

Any job should be able to run twice without duplicating data. In other words, jobs should be idempotent.

Look for a unique key in the source data and use it to avoid writing duplicate data.

For instance, you can set a Postgres unique index on the key. Then, use Rails' [create_or_find_by](https://blog.bigbinary.com/2019/03/25/rails-6-adds-create_or_find_by.html) method to create or update a record, depending on whether the key already exists.

### Use constraints to reject invalid data

Set constraints to stop invalid data from coming in. If a field should always have a value, use a [Postgres constraint](https://www.postgresql.org/docs/9.4/ddl-constraints.html) to require that field. Use Ruby's strict [conversion methods](https://avdi.codes/a-ruby-conversion-idiom/) to throw an error and abort a job if you're expecting one data type (say, an integer like 3) but get another (say, a string like "n/a").

Data is never perfect. If you catch invalid data errors early, you can adjust your pipeline to handle unexpected values.

What to read next
-----------------

A job management library like Sidekiq is critical to a data pipeline. Read more about [Sidekiq](https://github.com/mperham/sidekiq), the backbone of this example. Its [best practices docs](https://github.com/mperham/sidekiq/wiki/Best-Practices) are small and useful. If you're not using Ruby, find the top job management libraries for your language.

To practice creating a pipeline, find a public dataset that interests you and build a pipeline for it. The United States Government, for instance, lists its open government data on [data.gov](https://data.gov). Datasets range from [physician comparison data](https://data.medicare.gov/data/physician-compare) and a [database of every dam in the U.S](https://nid.sec.usace.army.mil/ords/f?p=105:19:7020640354040::NO:::).

Finally, check out managed solutions like [Google Dataflow](https://cloud.google.com/dataflow) for particularly large pipelines.

Have questions or want advice? Send me a tweet @coreybeep



----
Originally posted on [my blog](https://coreym.info/data-pipelines-in-ruby-on-rails/)

Photo by [Victor Garcia](https://unsplash.com/@victor_g?utm_source=unsplash&utm_medium=referral&utm_content=creditCopyText) on [Unsplash](https://unsplash.com/s/photos/pipeline?utm_source=unsplash&utm_medium=referral&utm_content=creditCopyText)