harvest.data.gov

harvest.data.gov is Data.gov's automated metadata ingestion service, built on the datagov-harvester application. It replaced the legacy CKAN-based harvester (H1.0) in 2025 -- this current system is referred to as Harvesting 2.0 (H2.0).

What it does

• Pulls metadata from agency harvest sources on a scheduled basis
• Supports DCAT-US JSON and WAF source types
• Manages harvest sources, jobs, and records for 120+ publishing organizations

Access

• Production: harvest.data.gov
• Source code: GSA/datagov-harvester

Technologies

• Python / Flask
• Postgres

Terminology

• record: a metadata file/set that describes a dataset
• catalog: a collection of records published together from a department/bureau/agency/program
• federal catalog: the Data.gov collection of metadata records across the federal government
• ETL: Extract, Transform, Load -- the framework used by the harvester
• source state: Data.gov's copy of the harvest source at a point in time (WAF, JSON, etc.)
• sourceId: UUID generated on creation of a new harvest source
• jobId: UUID generated when a new job is initiated
• recordId: UUID generated to track status of a record within the pipeline

Pipeline stages

Every harvest job runs through some or all of these stages, implemented as methods on HarvestSource in harvester/harvest.py:

1. EXTRACT -- fetch the source's current records (from a document URL or WAF listing) and the source's existing records from the harvest DB.
2. COMPARE -- hash each external record and compare against the stored source_hash for existing records to classify each as create / update / delete.
3. TRANSFORM (optional, non-DCAT sources only) -- for ISO 19115 sources, send the record to MDTranslator (a separate service, ghcr.io/gsa/mdtranslator in docker-compose.yml) to convert it to DCAT-US. DCAT sources skip this stage.
4. VALIDATE -- validate the (possibly transformed) record against the DCAT-US JSON Schema. Failures are logged as a harvest_record_error and the record is marked error_validation.
5. SYNC -- persist create/update/delete actions to the shared Postgres dataset table (HarvestRecord.sync()) and update the OpenSearch index accordingly. Failures are logged as a harvest_record_error and the record is marked error.
6. REPORT -- write job-level metrics (records added/updated/deleted/errored) to the harvest_job row and email a summary to the source's configured notification addresses.

Which stages actually run depends on the job type below -- e.g. a validate job runs EXTRACT/COMPARE/TRANSFORM/VALIDATE but skips SYNC.

sequenceDiagram
    autonumber
    participant FA as Flask App<br/>(LoadManager)
    participant HDB as Harvest DB<br/>(Postgres)
    participant TH as Task Handler<br/>(CF or Local)
    participant HS as Agency<br/>Harvest Source
    participant MDT as MDTranslator
    participant OS as OpenSearch
    participant SMTP

    note over FA: TRIGGER<br/>scheduled (source frequency) or manual (UI/API)
    FA->>+HDB: create harvest_job (status: new)
    HDB-->>-FA: returns harvest_job
    FA->>+TH: start_task: python harvester/harvest.py job_id job_type
    TH-->>-FA: returns OK
    TH->>HDB: update harvest_job status: in_progress

    note over TH: EXTRACT
    TH->>+HS: fetch source records (document/WAF)
    HS-->>-TH: source records
    TH->>+HDB: fetch existing records for source
    HDB-->>-TH: existing records

    note over TH: COMPARE
    loop hash each external record
        TH->>TH: classify create/update/delete via source_hash
    end

    note over TH: TRANSFORM (iso19115 sources only)
    loop non-DCAT records
        TH->>+MDT: translate to DCAT-US
        MDT-->>-TH: transformed record
    end

    note over TH: VALIDATE
    loop create/update records
        TH->>TH: validate against DCAT-US JSON Schema
        alt validation fails
            TH->>HDB: log harvest_record_error, status: error_validation
        end
    end

    note over TH: SYNC (skipped when job_type=validate)
    loop create/update/delete records
        TH->>HDB: upsert or delete dataset row
        TH->>OS: index or remove document
        alt sync fails
            TH->>HDB: log harvest_record_error, status: error
        end
    end

    note over TH: REPORT
    TH->>HDB: update harvest_job {status: complete, counts}
    TH->>SMTP: email job summary to source notification_emails

Loading

Job types

Set on HarvestJob.job_type (shared/constants.py, app/api/harvest_sources.py), a job is one of:

• harvest (default) -- the full pipeline above; only changed records (per COMPARE) are synced.
• force_harvest -- like harvest, but every record is treated as changed regardless of its hash, forcing a re-sync of the entire source.
• validate -- runs EXTRACT/COMPARE/TRANSFORM/VALIDATE only; used to check a source's data quality without writing anything.
• clear -- skips EXTRACT/COMPARE/TRANSFORM/VALIDATE and deletes every dataset for the source.

Jobs are triggered either automatically on the source's configured frequency (via LoadManager.schedule_next_job()), or manually via the UI/API (LoadManager.trigger_manual_job()), which is what the /api/harvest_source/harvest/<id>/<type> endpoint calls.

Source and schema types

• source_type: document (a single metadata file/URL), waf (a web-accessible folder listing many files), or waf-collection (a WAF whose records are children of a synthesized parent dataset -- see the Collections section on catalog.data.gov).
• schema_type: the metadata standard the source publishes in -- dcatus1.1: federal, dcatus1.1: non-federal, dcatus3.0, or an iso19115 variant. DCAT-US sources go straight to VALIDATE; ISO sources go through the TRANSFORM stage first.

Scheduling and job lifecycle

LoadManager (harvester/lib/load_manager.py) runs on every Flask Admin app startup, which cloud.gov triggers roughly every 15 minutes via a rolling restart:

1. Clean up -- any job still in_progress in the DB with no matching running CF task is marked error and reported.
2. Start new jobs -- up to HARVEST_RUNNER_MAX_TASKS concurrently running jobs are pulled from the queue of new jobs whose scheduled time has passed and started as Cloud Foundry tasks (or, locally, as a subprocess via LocalTaskHandler -- see docs/developer.md).

flowchart TD
    A[Flask Admin app starts<br/>~every 15 min via rolling restart] --> B{CF_INSTANCE_INDEX == 0?}
    B -->|no| Z[Skip -- only instance 0 runs this]
    B -->|yes| C[Clean up: get in_progress jobs from DB]
    C --> D[Get active CF tasks]
    D --> E{Job has a<br/>matching running task?}
    E -->|yes| F[Leave alone]
    E -->|no| G[Mark job error<br/>+ log harvest_job_error<br/>+ email failure notice]

    F --> H[Start new jobs]
    G --> H
    H --> I{Running tasks <<br/>HARVEST_RUNNER_MAX_TASKS?}
    I -->|no| J[Do nothing this cycle]
    I -->|yes| K[Pull 'new' jobs whose<br/>date_created has passed,<br/>up to available slots]
    K --> L[start_job: mark in_progress,<br/>launch CF task or LocalTaskHandler subprocess]
    L --> M[schedule_next_job:<br/>queue the source's next<br/>run per its frequency]

Loading

See Harvest O&M and Debugging for how to enable/disable harvesting fleet-wide and how to schedule large numbers of jobs without hitting the concurrency cap.

Related pages

• Harvest O&M and Debugging -- operational procedures

---

The sequence diagrams under docs/diagrams/mermaid/ in the datagov-harvester repo describe an earlier design that synced to CKAN (package_create/dataset_purge) and included job types (sync, retry) that no longer exist. The current pipeline syncs directly to Postgres and OpenSearch as described above -- treat those diagrams as historical, not current. Updated 2026-07-23.