Ingestion Pipelines

Hematite's ingestion system imports data from external sources — APIs, files, databases — into the substrate. Automated schema detection, field mapping, and validation against content type definitions ensure imported data integrates seamlessly.

Pipeline Stages

  1. SOURCE DETECTION — The pipeline identifies the source type (API endpoint, file upload, database connection) and extracts raw data.
  2. SCHEMA DETECTION — Automated analysis of the incoming data structure to infer field names, types, and relationships.
  3. FIELD MAPPING — Detected fields are mapped to target content type definitions. Unmapped fields can be stored as metadata or ignored.
  4. VALIDATION — Data is validated against the target schema: required fields present, types correct, constraints satisfied.
  5. TRANSFORMATION — Optional enrichment steps: language detection, entity extraction, sentiment analysis, cross-referencing against existing knowledge.
  6. PERSISTENCE — Validated data is stored in PostgreSQL (structured) or object storage (unstructured), with search indexes updated via CDC.
  7. PROVENANCE — An Ed25519-signed provenance entry records the origin, transformation steps, and current state of the ingested data.

Source Types

REST API Poll external REST endpoints, parse JSON/XML responses, map fields to content types
File Upload CSV, JSON, PDF, DOCX — text extraction for documents, structured parsing for CSV/JSON
Database Connect to external PostgreSQL, MySQL, or SQLite databases; query and import records
Web Scraping HTML parsing with configurable selectors, extract structured data from web pages
Webhook Receive real-time data pushes from external systems via webhook endpoints

Creating a Pipeline

// Define an ingestion pipeline
curl -X POST https://myorg.hematite.io/api/v1/ingest/pipeline \
  -H "Authorization: Bearer $TOKEN" \
  -d '{
    "name": "external-news",
    "source_type": "api",
    "config": {
      "url": "https://news-api.example.com/articles",
      "method": "GET",
      "headers": {"Authorization": "Bearer $API_KEY"},
      "pagination": {"type": "cursor", "param": "next_cursor"}
    },
    "target_type": "blog-post",
    "field_mapping": {
      "title": "$.title",
      "body": "$.content",
      "author": "$.author.name",
      "published_date": "$.date"
    },
    "schedule": "0 */6 * * *",
    "validation_rules": {
      "required_fields": ["title", "body"],
      "max_body_length": 50000
    }
  }'

Scheduled Ingestion

Pipelines can run on a cron schedule or be triggered manually:

// Trigger a pipeline immediately
curl -X POST https://myorg.hematite.io/api/v1/ingest/pipeline/{pipeline_id}/run \
  -H "Authorization: Bearer $TOKEN"

// View pipeline execution history
curl https://myorg.hematite.io/api/v1/ingest/pipeline/{pipeline_id}/history \
  -H "Authorization: Bearer $TOKEN"

// Update schedule
curl -X PUT https://myorg.hematite.io/api/v1/ingest/pipeline/{pipeline_id}/schedule \
  -H "Authorization: Bearer $TOKEN" \
  -d '{"cron": "0 */4 * * *"}'

Error Handling

Pipelines track failures and provide retry logic:

Connection Error Retry with exponential backoff (max 3 attempts). Alert on persistent failure.
Validation Error Record failed records with error details. Continue processing valid records.
Schema Mismatch Auto-detect new fields, create optional mappings, alert admin for review.

Key principle: Import data from any source, validate against your schema, enrich with agent-driven transformations, and publish — all with full provenance tracking from origin to current state.

Related Pages