I recently had a client hand me a folder full of CSV exports from an old legacy system. They wanted this data moved into a modern Postgres database, but looking at the files, it was a disaster: inconsistent date formats, random "N/A" strings where numbers should be, and a few rows that were just completely mangled. This is the classic ETL (Extract, Transform, Load) scenario.

Usually, when people first tackle this in Ruby, they try to do everything in one big loop. Let me show you where I started with this project and why it blew up in my face.

The "Just Load It All" Mistake

My first instinct was to keep it simple. I thought, "I'll just read the whole CSV, map over the rows, and save them." It looked something like this:

require 'csv'

def migrate_data(file_path)
  rows = CSV.read(file_path, headers: true)
  rows.each do |row|
    User.create!(
      email: row['Email'],
      joined_at: Date.parse(row['JoinDate'])
    )
  end
end

This worked for a file with 100 rows. But when I ran it against the real production export—which was about 400MB—my laptop's fans started screaming, and the process eventually crashed with a NoMemoryError. CSV.read loads the entire file into RAM. That's a rookie mistake I still make occasionally when I'm rushing. We need to stream the data instead.

Wrestling with Dirty Data

Once I switched to CSV.foreach to save my memory, I hit the second wall: the data itself. I started seeing ArgumentError: invalid date because some rows had dates like "Unknown" or "00-00-0000".

I could have just wrapped the whole loop in a begin...rescue block, but that's a nightmare for debugging. If one row out of a million fails, I don't want the whole pipeline to crash, but I also don't want to silently ignore the failure. I need a way to "clean" the data before it ever touches the database.

I decided to pull the transformation logic out into its own class. I wanted a dedicated object that takes a raw row and returns a "clean" hash, or tells me explicitly that the row is garbage.

class UserTransformer
  def transform(row)
    {
      email: row['Email']&.downcase&.strip,
      joined_at: parse_date(row['JoinDate'])
    }
  rescue StandardError => e
    # I'll return nil to signal this row is unprocessable
    nil
  end

  private

  def parse_date(date_str)
    return nil if date_str.nil? || date_str == 'Unknown'
    Date.parse(date_str)
  end
end

Building the Actual Pipeline

Now I have a way to extract (the CSV loop) and a way to transform (the Transformer class). But I'm still just calling User.create! inside a loop. In a real production environment, doing a database write for every single row is incredibly slow because of the overhead of individual transactions.

To fix this, I'll implement a "buffer." I'll collect transformed rows in an array and insert them in batches. This is where the "Load" part of ETL becomes efficient.

Here is how I structured the final version. Notice how each stage is decoupled. I can test the UserTransformer in isolation without even needing a CSV file.

require 'csv'

class UserETLPipeline
  BATCH_SIZE = 1000

  def initialize(file_path)
    @file_path = file_path
    @transformer = UserTransformer.new
    @buffer = []
  end

  def run
    CSV.foreach(@file_path, headers: true) do |row|
      clean_data = @transformer.transform(row)
      
      if clean_data
        @buffer << clean_data
      else
        puts "Skipping corrupt row: #{row.to_h}"
      end

      flush_buffer if @buffer.size >= BATCH_SIZE
    end
    
    flush_buffer # Don't forget the remaining rows!
  end

  private

  def flush_buffer
    puts "Loading #{@buffer.size} records into DB..."
    # In a real app, use User.insert_all(@buffer) for bulk insertion
    # User.insert_all(@buffer) 
    @buffer.clear
  end
end

By separating the concerns, I've turned a fragile script into a pipeline. If the business logic for "cleaning" a user changes, I only touch the transformer. If we switch from a CSV to a JSON API, I only change the extraction loop. The "Loading" logic stays the same regardless of where the data came from.

📋 Practical Task

Exercise: The Product Price Sync Pipeline

You've been given a CSV file named products.csv with the following columns: sku, price, and category. However, the data is messy: some prices contain currency symbols (e.g., "$19.99"), some are blank, and some categories are capitalized inconsistently.

Build an ETL pipeline that does the following:

• Extract: Read the products.csv file using a memory-efficient method.
• Transform: Create a ProductTransformer class that:
  • Removes the "$" symbol from the price and converts it to a Float.
  • Downcases the category name.
  • Returns nil if the SKU is missing.
• Load: Implement a ProductPipeline class that processes the data in batches of 5. Since you don't have a real database, your flush_buffer method should simply pp (pretty print) the array of cleaned hashes to the console to prove the batching is working.

Verification: When you run your pipeline, you should see the output printed in groups of 5, and any rows with missing SKUs should be skipped entirely.