How do I set up a source-to-warehouse pipeline for batch ingestion from APIs and SaaS tools?

A practical source-to-warehouse batch pipeline from APIs and SaaS tools usually looks like this:

1) Define the target architecture

Typical flow:

Source APIs / SaaS → Ingestion layer → Raw object storage or landing tables → Transformations → Analytics warehouse

Common choices:

• Ingestion: Airbyte, Fivetran, Meltano, custom Python jobs, Dagster, Airflow
• Landing zone: S3 / GCS / Azure Blob, or raw schema in the warehouse
• Warehouse: Snowflake, BigQuery, Redshift, Databricks SQL
• Transformations: dbt, SQL jobs, Spark, warehouse-native procedures
• Orchestration: Airflow, Dagster, Prefect, cron + scripts for simple setups

2) Decide ingestion pattern for each source

For batch ingestion, most APIs fall into one of these patterns:

A. Full refresh

Pull the entire dataset every run. Use when:

• Data volume is small
• The API doesn’t support incremental updates
• Simplicity matters more than efficiency

Pros: simple
Cons: expensive, slow, can hit rate limits

B. Incremental by timestamp

Pull only records changed since the last successful load. Use when:

• Source provides updated_at, modified_since, or similar

Store a watermark like:

• last successful updated_at
• last cursor/token
• last page processed

C. Incremental with change tokens or sync cursors

Some SaaS APIs provide:

• sync tokens
• delta links
• continuation tokens

This is ideal because it reduces backfill complexity and API usage.

D. CDC-like behavior for APIs

Not true CDC, but you can simulate it by:

• polling changed records
• upserting into warehouse
• tracking deletes if the API supports tombstones or deleted flags

3) Build a robust extraction layer

For each source, implement:

• Authentication: API key, OAuth, service account, etc.
• Pagination: offset/page-number/cursor
• Rate limit handling: exponential backoff, retries, sleep windows
• Incremental state: persist cursor/watermark in a state store
• Schema capture: raw JSON payload or flattened columns
• Error handling: retries for transient failures, dead-lettering for bad records
• Logging/metrics: records fetched, bytes, latency, failure counts

Good practice:

• Keep the raw payload unchanged in a landing table or object store
• Add metadata columns:
  • source_name
  • ingested_at
  • batch_id
  • load_mode
  • source_updated_at
  • record_hash

4) Land data in a raw layer first

Instead of writing directly into final warehouse models, land the data first.

Option 1: Raw files in object storage

Write JSON/CSV/Parquet to:

• s3://bucket/source/entity/dt=YYYY-MM-DD/part-*.parquet

Then load into the warehouse.

Option 2: Raw tables in the warehouse

Load into tables like:

• raw_salesforce.accounts
• raw_google_ads.campaigns

This is simpler if the warehouse is your central platform.

Best practice:

• Preserve source structure as much as possible
• Avoid heavy transformations during extraction

5) Normalize and model in the warehouse

After raw ingestion, transform into analytics-ready tables.

Common modeling layers:

• Staging: lightly cleaned, typed, standardized
• Intermediate: joins, deduplication, business logic
• Mart / dimensional models: facts and dimensions for reporting

Example:

• raw_hubspot.contacts
• stg_hubspot__contacts
• dim_customers
• fct_tickets

Use dbt if you want:

• SQL-based transformations
• version control
• tests
• documentation
• lineage

6) Handle upserts and deduplication

Most SaaS data is mutable, so you’ll need merge logic.

Patterns:

• Deduplicate by source ID + latest updated_at
• Merge into target tables
• Soft delete records when source marks them deleted
• SCD Type 1 for overwrite behavior
• SCD Type 2 when you need history

Example merge keys:

• source_id
• account_id
• composite keys if needed

If using dbt, implement incremental models with merge or insert_overwrite depending on warehouse support.

7) Manage schema drift

APIs change. Fields appear/disappear.

Strategies:

• Store raw JSON so you never lose fields
• Use automated schema evolution where supported
• Alert on schema changes
• Keep transformation models tolerant to missing fields
• Version your source contracts if possible

8) Orchestrate the pipeline

A batch pipeline should run on a schedule:

• hourly
• daily
• weekly

Orchestration steps typically:

1. Extract source data
2. Validate raw load
3. Load to warehouse/raw schema
4. Run transformations
5. Run tests
6. Publish/notify

Add dependencies between sources if needed:

• e.g. load accounts before transactions

9) Add data quality checks

At minimum, check:

• row counts
• freshness
• null/duplicate key checks
• referential integrity
• source vs target reconciliation

Tools:

• dbt tests
• Great Expectations
• Soda
• custom SQL checks

Example checks:

• source_id is unique in staging
• updated_at is not null
• row count didn’t drop unexpectedly
• freshness within SLA

10) Track state and retries carefully

For incremental ingestion, the state store is critical.

Store:

• last successful cursor/watermark
• job status
• batch ID
• source version
• retry count

Important:

• Only advance the watermark after a successful load
• Make the job idempotent so reruns don’t duplicate data
• If a batch partially fails, rerun safely

11) Secure secrets and access

Use:

• secret manager (AWS Secrets Manager, GCP Secret Manager, Vault)
• least-privilege access
• separate credentials per environment

Don’t hardcode:

• API keys
• warehouse passwords
• OAuth refresh tokens

12) Monitor and alert

Track:

• job success/failure
• freshness lag
• throughput
• API rate limit usage
• anomalies in row counts
• schema drift

Alert on:

• failed syncs
• stale data
• missing partitions
• large drops/spikes in volume

13) A simple implementation blueprint

If you want a straightforward setup:

Small-to-medium scale

• Airbyte for ingestion from SaaS APIs
• S3/GCS or warehouse raw schema for landing
• Snowflake/BigQuery as warehouse
• dbt for transformations
• Airflow/Dagster for orchestration and scheduling

Custom approach

• Python extractor using requests + pagination + retries
• Store state in Postgres or DynamoDB
• Write raw Parquet to object storage
• Load into warehouse with COPY/LOAD jobs
• Transform with dbt
• Orchestrate with Airflow

14) Example batch design

For an API like Salesforce or HubSpot:

1. Authenticate with OAuth
2. Query incremental records using updated_at > last_watermark
3. Handle pagination and rate limits
4. Write raw records to object storage
5. Load raw data into raw_salesforce.contacts
6. Run dbt staging model to cast fields and dedupe
7. Merge into dim_contacts
8. Run tests
9. Update watermark after success

15) Recommended best practices

• Prefer incremental over full refresh when possible
• Keep raw data immutable
• Make jobs idempotent
• Separate extraction from transformation
• Use warehouse-native incremental merges
• Add observability from day one
• Design for schema drift and API changes

If you want, I can also provide:

1. a reference architecture diagram,
2. a sample Airflow/Dagster/dbt setup,
3. or a Python example for ingesting a paginated API into Snowflake/BigQuery.