Mini GBIF Ingestion & Indexing Pipeline

This repository contains a runnable demo that ingests a subset of Norwegian GBIF datasets with an Airflow-orchestrated pipeline, validates / enriches the records in a Java Spark job, stores canonical results in Postgres + MinIO, indexes a search view in Elasticsearch, and exposes a FastAPI service over the processed data.

              ┌────────────────┐
              │ GBIF IPT (DwC) │
              └──────┬─────────┘
                     │ HTTP download
 ┌──────────────┐    ▼
 │   Airflow    │ fetch/upload ┌────────┐
 │ (scheduler)  │──────────────► MinIO  │ raw, processed Parquet
 └────┬─────────┘               └──┬─────┘
      │SparkSubmit                 │
      ▼                            │
┌──────────────┐  validations      │s3a://processed
│ Spark (Java) │───────────────────┘
└────┬─────────┘
     │   pandas/s3fs
     ▼
┌──────────────┐        ┌────────────────────┐
│ Postgres     │◄──────►│ FastAPI + Elastic  │
│ ingest schema│ meta   └────────────────────┘
└──────────────┘

Key Features

• Airflow DAG (airflow/dags/mini_gbif_ingest.py) orchestrates dataset fetch, Spark transform, Postgres load, and Elasticsearch indexing tasks for each configured dataset.
• Object storage first using MinIO buckets (raw, processed) with checksum-tracked objects for raw DwC-A payloads and canonical Parquet snapshots.
• Java Spark transform (spark-java/) enforces schema, normalizes values, applies TG2 validations (VALIDATION_COUNTRYCODE_STANDARD, VALIDATION_DECIMALLATITUDE_INRANGE), and writes valid/invalid splits plus summary stats.
• Relational store (Postgres schema ingest) keeps run metadata and a serving table (occurrence_valid) with indexes for queries / API usage.
• Search layer + API: Elasticsearch index occurrence backstops FastAPI endpoints for faceted search and record retrieval.
• Docker Compose environment brings up Airflow, Postgres, MinIO, Elasticsearch, and the FastAPI app locally.

Getting Started

1. Build the Spark jar (requires Maven + Java 17):

   ./scripts/build_spark.sh

2. Bring everything up:

   docker compose up --build

   • Airflow UI: http://localhost:8080 (user/pass: airflow / airflow)
   • MinIO console: http://localhost:9001 (user/pass: minio / minio123)
   • API service: http://localhost:8000/docs

3. Trigger the DAG inside the Airflow UI (mini_gbif_ingest), or via CLI once the containers are running:

   docker compose exec airflow-scheduler airflow dags trigger mini_gbif_ingest

4. Explore outputs:

   • Postgres (postgresql://pipeline:pipeline@localhost:5432/pipeline), schema ingest
   • Elasticsearch index occurrence via http://localhost:9200/occurrence/_search
   • API endpoints, e.g. GET /occurrence/search?countryCode=NO&basisOfRecord=PRESERVED_SPECIMEN

Configuration

Datasets live in config/datasets.yaml. Each entry supplies:

- dataset_key: <GBIF dataset UUID>
  dwca_url: <DwC-A download URL>
  publisher: <text>
  label: <human readable name>

Airflow mounts the file and creates per-dataset tasks automatically. Adjust, add, or remove datasets and re-trigger the DAG.

Pipeline Stages

1. Fetch + land raw data
   Downloads each DwC archive, extracts occurrence.txt, computes SHA-256, and stores it in MinIO under raw/datasetKey=<id>/ingestDate=<date>/. Metadata (bucket key, checksum, ingest date) rides through XCom to downstream tasks.

2. Spark transform (Java)
   Spark 3.5.1 job (MiniGbifTransform) runs on spark-submit --master local[*] inside the Airflow image with Hadoop S3A configured for MinIO.

   • Normalizes columns (countryCode, basisOfRecord, coordinates)
   • Validates ISO country codes (VALIDATION_COUNTRYCODE_STANDARD) + latitude range (VALIDATION_DECIMALLATITUDE_INRANGE)
   • Adds derived fields (event_year, lat_bucket, lon_bucket, taxon_key_hash)
   • Splits datasets into valid / invalid Parquet + stats JSON at s3a://processed/datasetKey=<id>/run_id=<ts>/

3. Load to Postgres
   pandas + sqlalchemy read the processed Parquet, insert run metadata into ingest.ingest_run, and bulk load valid rows into ingest.occurrence_valid (with indexes on (dataset_key, event_year) and (country_code, basis_of_record)).

4. Index Elasticsearch
   Each run re-indexes the dataset into the occurrence index with a simple mapping (text search on scientific_name, keyword facets, geo point from lat/lon).

5. API Service (api/app/main.py)
   FastAPI exposes:

   • GET /health
   • GET /occurrence/{gbif_id} (Postgres-backed lookup)
   • GET /occurrence/search (query/filters/bbox via Elasticsearch)
   • GET /datasets/{dataset_key}/runs (run history from Postgres)

Validations & Enrichment

• VALIDATION_COUNTRYCODE_STANDARD: enforce ISO 3166-1 alpha-2 (plus GBIF’s XZ, ZZ) via Spark UDF and route failures to the invalid Parquet with explicit issue codes.
• VALIDATION_DECIMALLATITUDE_INRANGE: ensures decimalLatitude ∈ [-90, 90]; a longitude sanity check is included as a bonus.
• Derived fields: event_year, coordinate buckets, SHA-256 hash over the scientific name to mimic stable partition keys.

CI / Automation Notes

• Add GitHub Actions to run mvn -q -pl spark-java test package and lint the Python pieces (ruff, pytest) to signal engineering rigor.
• Docker Compose keeps everything local, but infra/k8s/ could host manifests or Helm charts translating these services to AKS/EKS/GKE once ready (explain in your application how you would externalize MinIO → S3 and Postgres → managed service).

Next Steps / Scaling Notes

1. Swap MinIO with managed S3/GCS and wire S3 notifications → Airflow datasets for incremental triggers.
2. Extend the Spark job with additional TG2 validations (e.g., longitude range, month/day normalization) and partitioned Parquet layout for Trino/Arrow.
3. Add Observability: structured logs (already in Airflow), metrics via Prometheus, dashboards showing run stats from ingest.ingest_run.
4. Package the API + Elasticsearch as a tiny DigitalOcean deployment (docker compose → doctl apps) pointing to the remote Postgres read replica to mimic the target architecture.