Data Pipelines

In the previous sections, you learned fundamental techniques to build datasets.

By now you may have found an undocumented API, crowdsourced receipts, or used browser automation to collect web pages to use for an investigation. But how will you actually implement the data collection so you have enough data to stand up a claim?

What happens when a script crashes, and how do you keep track of your workflow when you have multiple dirty datasets involved?

Or perhaps my own nightmare, re-running a script from months ago returns a drastically different statistic. Never again!

In this section you’ll learn how to build data pipelines and best practices for data engineering.

👉 Click here to jump to the tutorial.

Intro

What is a data pipeline?

We’ll walk through a hypothetical task to introduce the best practices of a data pipeline.

We want to build a pipeline for courtlistener to download court opinions and analyze all the cases against a major business. In this case, courtlistener has an well-documented and free API. Take a moment to review the documentation.

A data pipeline is composition of narrowly-defined scripts.

Rather than having a monolithic script, it’s best to break up downloading, processing, and analysis.

The main benefits of data pipelines lie in proceduralizing a sequence of decisions:

1. Resilience – when a stage crashes, only that stage restarts. Everything already written is preserved.
2. Parallelism – running multiple instances of a script can receive inputs and run simultaneously. This allows for scale.
3. Reproducibility – every output is a file. Re-run any stage at any time from its inputs and get the same result.
4. Auditability – every process is programatic and self-documented.

Good engineering is about writing D.R.Y. code (Do not repeat yourself). When it comes to data pipelines, this means you only download something ONCE. For that reason being thoughful about how you organize directories and file names is key.

Compare this to a monolithic script that does everything in one pass: a network error at step 40,000 of 50,000 means losing all progress, and it is hard to test one part without running the whole thing.

Principled Data Processing

Patrick Ball a statistician and human rights data scientist, developed a framework for trustworthy data work: every processing step should be a self-contained, repeatable unit with well-defined inputs and outputs.

Immutability

Once input data is written, it is never modified. Every transformation (such as filtering, labelling, merging) should be done programatically and produce new output files.

These steps preserve the past and make sure that we never alter evidence that our analysis relies on.

Idempotency

Running a stage twice produces the same result as running it once.

This is where deliberate filenaming is key. The unique identifier used to collect the data should be encoded in the file name.

If the contents of that file doesn’t change, you should check if the file exists before doing anything. Conversely, if the data changes over time, choose to include the collection date into the file name and save a new immutable version.

These steps make sure repeating the same operations reproduces the exact same outputs.

Modularity

Each part of the data pipeline is narrowly-scoped to be responsible for the smallest-possible unit of work. This allows better fault-tolerance as well as scalabity. Certain components of a scraper can be slower than others. By dividing the tasks, you can better identify bottlenecks to optimize the pipeline.

As Patrick Ball tells us:

The idea is that each component of work we need to do should be as small and simple as possible. To accomplish a larger task, we link one piece to another. Keeping pieces small makes them simpler to design and debug, and using simple interfaces enables us to connect pieces together to build bigger structures.

Big problems should be be broken up into smaller problems. Being smart can help you get to the finish line faster and debug issues quicker.

Pro tip:

Keep each script’s entry point simple: one function call. This makes it easy to run stages manually, from a Makefile, or from an orchestrator without any special framework.

Dividing each stage of the pipeline into modular scripts helps orchestrate the task. Each script looks for the output of the previous step and checking against the work that has already been done at the present stage of the pipeline.

Stay Organized

Organization is everything. When building a data pipeline on a local machine (like your laptop), have a dedicated data directory.

No matter where you save your files, clearly delinate what is unprocessed data, derivied data, and manually-labelled data.

Utilize subdirectories to differentiate input for raw records, output for files we use for the analysis, and hand for anything manual. For example, see this abbreviatied directory for the “Still Loading” investigation into internet pricing disparities.

data
├── input
│   ├── redlining
│   ├── addresses
│   ├── census
│   └── isp
│       └── att
├── intermediary
│   ├── census
│   └── isp
│       └── att
└── output
    ├── figs
    └── tables

If using cloud storage, you can omit the data directory in lieu of a unique URL (or S3 bucket) for the project. The principals of separating input, output, and manual data remains of course.

As mentioned previously file names should be deterministic, allowing for quick reference to what has been collected.

Keeping the raw evidence (input) immutable means that reference stays trustworthy — what you collected doesn’t silently change later. If your pipeline is reproducible, your results could be deleted and regenerated based on the input files.

On Directory Structure:

See more examples of how directories were organized to investigate GPT bias and Amazon private label

Steps to develop data pipelines

A rough outline of how I approach building a data pipeline:

1. Identify the mechanism for data collection. Where are you getting data from?
2. Determine the file structure for storage. How are you saving the data and naming files?
3. Determine the scope. How much data is needed and how often will you collect it?
4. Build a trial version of the pipeline locally. Focus on a small subset. What are bottlenecks?
5. Parse and analyze the trial data. What can you claim with existing data?
6. Optimize the pipeline for “production”. How to make the pipeline scalable and robust?
7. Monitor progress. How to keep track that data is flowing and get a time estimate?
8. Debug. What are edge cases and how to make sure the dataset is complete and accurate?

Tutorial

Building a pipeline to collect federal court records

In Franz Kafka’s “The Trial”, the protagonist Josef K. is arrested without knowing the charge and forced through a non-sensical maze of bureaucracy.

Let’s instead make sense of the courts by building a principled data pipeline for the courts.

We will walk through a five-stage pipeline that searches CourtListener for court cases involving a company, then downloads case metadata and judge opinions – saving everything to either locally or to Amazon’s cloud storage S3.

The full code lives in this GitHub repository. This tutorial focuses on the key patterns – callouts throughout link to the exact function in the real script.

To follow along locally, clone the repository and set STORAGE_BACKEND=local in your .env. All files will be written to ./data/ instead of S3 – you will still need an AWS account to use SQS.

1. See you in court

CourtListener is a free, public database of U.S. federal court records maintained by the Free Law Project. They were kind enough to make the database accessible via a REST Application Programming Interface (API) at https://www.courtlistener.com/api/rest/v4/. Review the documentation before proceeding.

Court case data is stored in a multi-level hierarchy:

Docket (the case with parties, court, filing dates)
  └─ Cluster (the court decision of the case)
        └─ Opinion (one judge's opinion)

This hierarchy forces us to implement a multi-stage data pipeliine. We can’t get an opinion without first knowing the corresponding cluster ID, and subsequently cluster IDs are found in the docket metadata. So naturally, we need to fetch the first docket, then the cluster, and finally each judge’s opinion.

2. Naming files

As covered above, deterministic filenames are what make is_file_saved() possible.

Here’s what each type of file that looks like for this specific pipeline:

{company}/{docket_id}/docket-meta-{docket_id}.json
{company}/{docket_id}/clusters/{cluster_id}/cluster-meta-{cluster_id}.json
{company}/{docket_id}/clusters/{cluster_id}/opinions/{opinion_id}.json

For example, a Boeing docket with ID 12345 saves to:

Boeing/12345/docket-meta-12345.json
Boeing/12345/clusters/67890/cluster-meta-67890.json
Boeing/12345/clusters/67890/opinions/111213.json

Where:
67890 = the cluster ID
111213 = the opinion ID

The resulting directory would be organized like this:

Boeing/
└── 12345/
    ├── docket-meta-12345.json
    └── clusters/
        └── 67890/
            ├── cluster-meta-67890.json
            └── opinions/
                └── 111213.json

Filenames can be declared programmaticly using f-strings or filesystem software such as os.

Keeping track of the files is much like a save room in a video game: if your script “dies” partway through, re-running it doesn’t send you back to level one, it picks up from the last checkpoint you already wrote to disk.

Pro tip:

Encode the entity ID in the filename itself (docket-meta-12345.json rather than meta.json). Files become self-describing when browsing for files in cloud storage of when it’s saved locally. It also makes name collisions much less likely.

3. Architecture

Each of these different file types is generated from a separate script calling a different API endpoint. The pipeline begins by making a search for dockets IDs based on company name, collecting the docket metdata, listing the cluster metadata for each docket ID, then fetching the opinions for each cluster.

 {company}
     |
     V
[01_search.py]
     |
{docket_id}
     |
     V
[02_get_docket.py]  -->  {company}/{docket_id}/docket-meta-{docket_id}.json
     |
{cluster_id}
     |
     V
[03_get_clusters.py]  -->  {company}/{docket_id}/clusters/{cluster_id}/cluster-meta-{cluster_id}.json
     |
{opinion_id}
     |
     V
[04_get_opinions.py]  -->  {company}/{docket_id}/clusters/{cluster_id}/opinions/{opinion_id}.json

However, how do you keep state between each script? Especially as each step is dependent on the subsequent step.

Also which records change and which do not change?

4. Keeping track of what’s done

If you want to be minimalist, you can list and parse the contents of the preceding directory in order to get a list of IDs to fetch minus the IDs that are already in the output directory.

Here’s psuedocode for the minimalist approach:

import glob

def get_cluster_id(filename) -> str:
    return filename.split('-')[-1].split('.json')[0]

# list the docket and cluster files to determine which clusters collected
collected_dockets = glob.glob('data/input/*/*/docket-meta-*.json')
collected_clusters = glob.glob('data/input/*/*/clusters/*/cluster-meta-*.json')

# Gather cluster_ids into a set.
cluster_ids = set() 
for fn in collected_dockets:
    docket_meta = json.load(open(fn))
    cluster_ids.update(docket_meta.get('cluster_ids'))

# isolate cluster IDs using "set comprehension" (look this up)
collected_cluster_ids = {get_cluster_id(fn) for fn in collected_clusters}

# omit cluster IDs we already collected
clusters_to_collect = cluster_ids - collected_cluster_ids

However, this process scales poorly if you’re trying to collect a large dataset. Instead, we can use an intermediary to keep track of what needs to be done in each step.

Amazon Simple Queue Service (SQS) is a managed message queue – a holding area for tasks that workers pull from one at a time. The costs tend to be manageable, and the service works on any infrascture with internet access.

A queue is much like a ticketing system in a commercial kitchen. The server writes records orders in a point of sale system. The kitchen prints the ticket and adds it to the queue. When the order is fullfilled, the kitchen discards the ticket when the work is done. If the kitchen is interrupted, the ticket remains in the queue and another member of the kitchen picks it up.

A specialized order such as a drink would go to the bar’s queue whereas food goes to the kitchen’s queue.

In our case, rather than orders for meals we’ll be sending IDs for dockets, clusters, and opinions to collect. For each type of data, we will use a unique queue and separte script: - 01_search.py Inputs a company name and outputs docket IDs into the dockets queue for each new case found. - 02_get_docket.py Inputs docket IDs from the dockets queue. Collects docket metadata, and outputs corresponding cluster IDs into the clusters queue. - 03_get_clusters.py Inputs cluster ID from the cluster queue. Collets cluster metadata, and outputs opnion IDs into the opnions queue. - 04_get_opinions.py Inputs opinion ID from the opinion queue and collects the full text of the opinion.

The key rule: a message is deleted only after the work succeeds. If a worker crashes before deleting the message, SQS makes it visible again. So the last step of each script iteration is deleting the ticket from the SQS queue.

All four worker scripts are initiated by listing what tickets are in the queue via a generic poll_queue function that basically inputs a SQS message to any arbitrary process_func:

def poll_queue(queue_url, process_func):
    """
    Poll an SQS queue until empty, calling `process_func` for each message.
    Process function contains logic for each API (search/docket/cluster/opinion).
    Deletes message only after successful processing.
    """
    while True:
        messages = sqs.receive_message(
            QueueUrl=queue_url,
            WaitTimeSeconds=10
        ).get('Messages', [])
        
        if len(messages) == 0:
            print('Queue empty, exiting')
            break

        for message in messages:
            process_func(message)
            sqs.delete_message(    
                QueueUrl=queue_url,
                ReceiptHandle=message['ReceiptHandle'],
            )

Simplified psuedocode – the real poll_queue in utils.py also logs each message and pauses briefly between them.

5. The start of the pipeline

NOTE: lost steam here to be honest.

01_search.py is the first step of the data pipeline. It reads a state.json file that lists which companies to search and when each was last run, calls the CourtListener search API, and enqueues any docket IDs not yet collected.

One of the best takeaways from Patrick Ball’s principaled data processing is keeping manual data (such as the companies we’re interested) in a hand directory. Files stored elsewhere should be derived programatically.

When collecting a dataset a basic question is whether we’re looking for a snapshot of data or a continuous collection. Further we need to assess which files are static and which can change over time. For continuous data collection, consider how often the script runs. This all helps us keep track of what’s done and what needs to be done.

In the case of CourtListener, new dockets emerge daily and existing cases can have periodic updates.

Two things make the search incremental: - last_run – if set, the search only fetches cases filed after that date. On the first run it is null, which fetches all history up to cutoff_date. - is_file_saved() – within each run, it skips any docket whose output file already exists.

State is written back to (locally or to S3) only after all companies complete successfully. A partial run leaves last_run unchanged, so the next run re-fetches the same date range.

def search_company(query, state):
    last_run = state.get(query, {}).get('last_run')   # None on first run
    params = {'q': query, 'type': 'd'}
    if last_run:
        params['filed_after'] = last_run

    page_url = f'{CL_BASE_URL}/search/'
    while page_url:
        response = cl_session.get(page_url, params=params)
        data = response.json()

        for docket in data['results']:
            docket_id = docket['docket_id']
            fn_out = f'{query}/{docket_id}/docket-meta-{docket_id}.json'
            if is_file_saved(fn_out):
                continue   # already collected
            sqs.send_message(
                QueueUrl=DOCKETS_QUEUE_URL,
                MessageBody=json.dumps({'docket_id': docket_id, 'query': query}),
            )

        page_url = data.get('next')   # cursor-based pagination
        params = {}                   # next URL already contains params

    return {'last_run': date.today().isoformat()}


def search_all():
    state = read_state()
    for query in state:
        state[query] = search_company(query, state)
    write_state(state)   # write only after all companies succeed


if __name__ == '__main__':
    search_all()

Simplified for teaching – the real search_company/search_all also handle pagination edge cases, cutoff_date, and per-court filtering. The is_file_saved, read_state, and write_state helpers it calls all live in utils.py.

6. Worker scripts

Scripts 2, 3, and 4 all follow the same pattern: pull a message from a queue, fetch from the API, save the result, enqueue downstream work, delete the message. They run until the queue is empty, then exit.

02_get_docket.py demonstrates the pattern. It also handles the one exception to full immutability: dockets can gain new clusters as court decisions are filed, so the docket file is re-fetched once per day using is_file_from_today(). Cluster and opinion files are fully immutable once written.

The rate limiting policy is also worth noting: an HTTP 429 (rate limit) causes the worker to exit immediately via SystemExit, so the orchestrator can scale down on its next run. Transient server errors (502, 503, 504) are retried automatically with exponential backoff.

def fetch_and_save_docket(message):
    body = json.loads(message['Body'])
    docket_id = body['docket_id']
    query = body['query']
    fn_out = f'{query}/{docket_id}/docket-meta-{docket_id}.json'

    if is_file_from_today(fn_out):
        docket = json.loads(get_file(fn_out))   # re-use today's fetch
    else:
        response = cl_session.get(f'{CL_BASE_URL}/dockets/{docket_id}/')
        docket = response.json()
        put_file(fn_out, json.dumps(docket))

    # enqueue any clusters not yet saved
    for cluster_url in docket.get('clusters', []):
        cluster_id = cluster_url.rstrip('/').split('/')[-1]
        fn_cluster = f'{query}/{docket_id}/clusters/{cluster_id}/cluster-meta-{cluster_id}.json'
        if not is_file_saved(fn_cluster):
            sqs.send_message(
                QueueUrl=CLUSTERS_QUEUE_URL,
                MessageBody=json.dumps({'cluster_id': cluster_id, 'docket_id': docket_id, 'query': query}),
            )


if __name__ == '__main__':
    poll_queue(DOCKETS_QUEUE_URL, fetch_and_save_docket)

Simplified for teaching – the real fetch_and_save_docket also raises on HTTP errors and prints progress. The helpers it calls – is_file_from_today, get_file, put_file, and is_file_saved – all live in utils.py.

7. Monitoring

The quickest way to check our pipeline’s health is the SQS console. Look at NumberOfMessagesDeleted per minute – this is real-time throughput. A flat line means workers have stopped; a spike means workers came online and are draining the queue.

The 00_status.py script provides a local snapshot:

python 00_status.py

This prints: - Messages available in each queue (backlog) - Messages deleted in the last 15 minutes (throughput) - Total dockets, clusters, and opinions saved to S3 - Path of the most recently written file per type

Pro tip:

If the opinions queue is draining but the clusters queue is growing, you have a bottleneck at the cluster stage. Lower MESSAGES_PER_WORKER_CLUSTERS to launch more workers there sooner, and raise MESSAGES_PER_WORKER_OPINIONS to let opinion workers handle more work before another is added.

8. Docker and the Makefile

Packaging the pipeline in Docker means the same code runs identically on a laptop and in the cloud. One Dockerfile packages the entire codebase; all five stages run from the same image, differentiated only by the command in their Elastic Container Service (ECS) task definitions.

A Makefile collects the common operations (read more about Makefiles):

make push              # build the Docker image and push it to ECR
make register-tasks    # create one ECS task definition per stage
make run TASK=search   # launch one Fargate task manually
make dry-run           # preview orchestrator decisions without launching

Task definitions are immutable snapshots. If you change an environment variable – say, a new queue URL – re-run make register-tasks to create a new revision pointing at the updated value.

TODO: explain the AWS pieces before this point

This section currently name-drops several AWS services without defining them. Add a short paragraph here (mirror the Docker treatment above – a sentence or two plus a link out) covering:

• ECS (Elastic Container Service) – runs your Docker containers for you; you don’t manage the underlying servers.
• Fargate – the “serverless” compute mode for ECS. AWS provisions the machine per task instead of you managing EC2 instances.
• Fargate Spot – interruptible, discounted capacity. AWS can reclaim a task mid-run, which is fine here since a killed task’s SQS message just becomes visible again (ties back to the “Resilience” pillar from the intro).
• Task vs. task definition – a task definition is the blueprint (image, CPU/memory, command, env vars); a task is one running instance of it.
• EventBridge – AWS’s scheduler, mentioned in Section 10; what actually triggers 05_orchestrate.py on the cron expression.
• ECR and Secrets Manager – named later in this tutorial (make push, Section 9) with no explanation either.

Also flag the naming collision: “cluster” means an ECS cluster (a pool of Fargate capacity) and a CourtListener cluster (a court decision) throughout this tutorial – worth calling out explicitly so students don’t conflate the two.

Most of this is already spelled out in the repo’s README – the AWS setup table defines the ECS cluster, ECR repo, IAM roles, VPC/subnet, and security group in one place, and ECS deployment covers task definitions and the Makefile targets in depth. Students still need a short in-notebook explanation per the “not amazing coders” bar, but the README section can carry the full depth and setup steps – link to it rather than re-deriving.

9. Secrets and environment variables

All credentials live in a .env file that is listed in .gitignore. Never commit it.

cp .env_template .env
# fill in your values, then load them:
set -a; source .env; set +a

The pipeline needs four categories of values:

Category	Variables
From your instructor	AWS_DEFAULT_REGION, SUBNET_ID, SECURITY_GROUP_ID
Your AWS credentials	AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
Your resources	S3_BUCKET, DOCKETS_QUEUE_URL, CLUSTERS_QUEUE_URL, OPINIONS_QUEUE_URL
Your identity	TASK_PREFIX (prefixes ECS task names to avoid conflicts with classmates)

See the README’s student setup section for exactly where each value comes from and how to create the underlying S3 bucket and SQS queues, and .env_template for the full variable list.

In production, containers read credentials from AWS Secrets Manager – values are injected as environment variables at container startup. Nothing sensitive is baked into the Docker image.

Set STORAGE_BACKEND=local during development. All files will write to ./data/ and no S3 or SQS access is needed to test the core logic.

10. Orchestration and scheduling

For scripts that you want to schedule to run periodically on a laptop or local machine you can use crontab.

To manage how many tasks to run for each part of the pipeline, we can use 05_orchestrate.py. It checks how many messages are waiting in each queue and how many workers are already running, then launches additional Fargate Spot workers to close the gap.

The scaling formula for each stage:

desired workers = ceil(queue depth / MESSAGES_PER_WORKER)
workers to launch = min(budget_remaining, desired - running)

MAX_WORKERS (default: 5) caps total running tasks across all stages – the primary cost control. MESSAGES_PER_WORKER is a per-stage tuning knob: lower values launch workers more aggressively; higher values let each worker drain more before another starts. See the README’s ECS deployment section for worked examples of how queue depth translates into worker counts.

The orchestrator runs on a cron schedule via AWS EventBridge. A cron expression like */10 * * * * runs it every 10 minutes. crontab.guru is a useful reference for writing and reading these expressions.

Use --dry-run to preview scaling decisions without launching anything – useful for tuning MESSAGES_PER_WORKER values before spending money:

def main(dry_run=False):
    running_counts = {family: get_running_count(family) for family, _ in STAGES}
    total_running = sum(running_counts.values())
    worker_budget = MAX_WORKERS - total_running

    for family, queue_url in STAGES:
        depth = get_queue_depth(queue_url)
        running = running_counts[family]
        desired = math.ceil(depth / MESSAGES_PER_WORKER[family]) if depth else 0
        to_launch = min(worker_budget, max(0, desired - running))
        worker_budget -= to_launch

        print(f'{family}: queue={depth:,}  running={running}  launching={to_launch}')
        if to_launch > 0 and not dry_run:
            launch_tasks(family, to_launch)


if __name__ == '__main__':
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument('--dry-run', action='store_true')
    main(dry_run=parser.parse_args().dry_run)

Simplified for teaching – the real main in 05_orchestrate.py snapshots running counts across all stages before deciding how much of the worker budget each one gets, so an earlier stage can’t starve a later one. It calls get_queue_depth, get_running_count, and launch_tasks, all defined in the same file.

Running it yourself

Local development

1. Install dependencies

pip install -r requirements.txt

2. Configure your environment

cp .env_template .env
# set STORAGE_BACKEND=local and fill in your CourtListener API key
set -a; source .env; set +a

3. Create state.json

Create a file called state.json in the project root with the companies you want to search:

{
  "Boeing": {
    "last_run": null,
    "cutoff_date": "2020-01-01"
  }
}

last_run: null means the first run fetches all history up to cutoff_date. Subsequent runs only fetch cases filed since the last run.

4. Run the stages in order

python 01_search.py       # search and enqueue docket IDs
python 02_get_docket.py   # fetch docket metadata
python 03_get_clusters.py # fetch cluster records
python 04_get_opinions.py # fetch full opinions

Each script exits when its queue is empty. Check progress at any point:

python 00_status.py

AWS deployment

make push              # build and push Docker image to ECR
make register-tasks    # register ECS task definitions
make run TASK=search   # kick off the search stage
make orchestrate       # scale workers based on queue depth

See the README for full AWS setup, including the one-time instructor setup (infra/setup.sh) and per-student resource creation.

Exercises

Practice: Add a second company to state.json – try a company currently in the news for legal reasons – and run the pipeline locally. How many dockets does it find? Browse ./data/ and describe the file structure. What does the JSON in a docket metadata file contain?

Tuning: Modify MESSAGES_PER_WORKER_DOCKET in your .env and run python 05_orchestrate.py --dry-run to preview how the scaling decisions change. What happens when you set it to 1? To 500?

Extend the pipeline: Add a new script 06_extract.py that reads opinion JSON files from ./data/, extracts the plain text of the opinion (it is in the plain_text field), and saves it as a .txt file alongside the original JSON. Guard every write with is_file_saved() so the script is idempotent.

Homework assignment

Scoping: Identify a company or topic you want to investigate using federal court records. Set a cutoff_date that bounds the collection to a manageable time range. Run a trial collection locally and inspect 10 dockets manually – what kinds of cases appear? What fields in the docket metadata might be useful for filtering or categorization?

Reporting: What questions can you answer with the opinions you collected? Which courts appear most frequently? Which clusters have multiple opinions (indicating appeals or dissents)? What would you need to answer a concrete investigative question with this data?

Citation

To cite this chapter, please use the following BibTeX entry:

@incollection{yin2026pipelines,
  author    = {Yin, Leon},
  title     = {Data Pipelines},
  booktitle = {Inspect Element: A practitioner's guide to data-driven investigations},
  year      = {2026},
  editor    = {Yin, Leon and Sapiezynski, Piotr},
  note      = {\url{https://inspectelement.org}}
}

Acknowledgements

Thanks to students in the Frontiers of Computational Journalism course (Spring 2026) for testing this pipeline and improving the setup documentation.