Get Started
Get Started

How To Ingest CRM Data Into a Data Lake: A Guide

How To Ingest CRM Data Into a Data Lake: A Guide

Content

Coffee is the
AI-first CRM

Get Started

Written by: Doug Camplejohn, CEO & Co-Founder, Coffee

Key Takeaways

  • CRM data in Salesforce, HubSpot, and Dynamics 365 stays siloed by default, so you need explicit pipelines to feed your data lake.
  • Manual CSV exports create stale, error-prone data. Automated pipelines using Bulk API, CDC, or managed connectors keep data fresh and reliable.
  • Raw landing zones should store untouched source data with Hive-style partitioning and metadata fields such as ingested_at, batch_id, and source_system.
  • Incremental loads with watermark columns and SCD Type 2 handling keep pipelines efficient while preserving history for objects like Accounts and Contacts.
  • Coffee improves CRM data quality at the source, which reduces the need for complex extraction workarounds and keeps clean data flowing into your lake.

Explore Coffee Pricing

Why Reliable CRM Ingestion Pipelines Matter

Coffee’s Market Data shows that 71% of sales reps spend too much time on data entry, leaving only 35% of their time for actual selling. That manual burden degrades CRM data quality at the source. A well-designed ingestion pipeline enforces good data in and good data out. Structured extraction, schema-aware landing, and incremental loading keep the lake aligned with real-world activity instead of partial notes. Complex CRM object models, careful rate limit handling, and SCD Type 2 history logic together explain why a production-grade pipeline often takes days to weeks to build and test.

Readiness And Preconditions

Confirm a few basics before you start writing extraction code:

  • API access credentials for your CRM: a Connected App with JWT Bearer OAuth 2.0 for Salesforce, a Private App access token for HubSpot (legacy API keys were sunset in November 2022), or a service principal for Dynamics 365
  • A target data lake such as S3, ADLS Gen2, or GCS with appropriate IAM roles and bucket policies
  • An orchestration tool such as Apache Airflow, Azure Data Factory, or Databricks Lakeflow Jobs
  • A basic understanding of ELT patterns: load raw data first, then transform inside the warehouse or lake

Step 1: Map The CRM Object Model And Initial Scope

Every major CRM organizes data around a common set of core objects. Start with these before moving to custom objects:

  • Accounts: the company or organization record; primary key is AccountId in Salesforce
  • Contacts: individual people associated with an Account
  • Leads: unqualified prospects not yet linked to an Account
  • Opportunities: active deals with stage, amount, and close date
  • Cases: support tickets linked to Contacts or Accounts

Map the foreign key relationships before extraction. Contacts have an AccountId foreign key. Opportunities have both AccountId and ContactId. Cases link back to both. These relationships determine join order in the silver layer and which objects must land first so others can be enriched.

Step 2: Choose Your Extraction Method

The extraction method sets your throughput, latency, and operational complexity. Use the table below to compare the four main options by ideal use case and the primary limit you will hit first, then match the method to your volume and freshness needs.

Method Best For Key Limit
REST API Low-volume, real-time lookups 100,000 + (1,000 × license count) calls/24 hrs (Salesforce Enterprise)
Bulk API 2.0 High-volume batch extraction 150 MB CSV per upload chunk; 150M records/24 hrs
CDC / Streaming Near-real-time delta capture 3-day (72-hour) event replay window (Salesforce Pub/Sub API)
Managed Connector Speed to value, low maintenance Vendor-managed; schema drift handled automatically

HubSpot rate limits vary by tier. Free/Starter allows 100 requests per 10 seconds and 250,000 per day; Professional allows 190 requests per 10 seconds and 650,000 per day; Enterprise allows 190 requests per 10 seconds and 1,000,000 per day. HubSpot’s CRM Search API often becomes the bottleneck. It is capped at 4 requests per second and 10,000 total results per query. Prefer list endpoints with the after cursor token for full exports.

The Dataverse Web API for Dynamics 365 returns up to 5,000 rows per request and uses server-driven paging via @odata.nextLink. Offset-based pagination does not work here, so plan for cursor-based loops.

The following Python snippet submits a Salesforce Bulk API 2.0 query job and polls for results:

import requests, time SF_INSTANCE = "https://yourorg.my.salesforce.com" ACCESS_TOKEN = "YOUR_ACCESS_TOKEN" headers = { "Authorization": f"Bearer {ACCESS_TOKEN}", "Content-Type": "application/json", } # Create query job job_payload = { "operation": "query", "query": "SELECT Id, Name, LastModifiedDate FROM Account", "contentType": "CSV", } job_resp = requests.post( f"{SF_INSTANCE}/services/data/v60.0/jobs/query", json=job_payload, headers=headers, ) job_id = job_resp.json()["id"] # Poll until complete while True: status = requests.get( f"{SF_INSTANCE}/services/data/v60.0/jobs/query/{job_id}", headers=headers, ).json()["state"] if status == "JobComplete": break time.sleep(10) # Download results (paginate via locator) locator = None while True: url = f"{SF_INSTANCE}/services/data/v60.0/jobs/query/{job_id}/results" if locator: url += f"?locator={locator}" result = requests.get(url, headers=headers) # Write result.text (CSV) to landing zone locator = result.headers.get("Sforce-Locator") if not locator or locator == "null": break

Step 3: Design The Raw Landing Zone

The raw landing zone acts as the source of truth for your pipeline. Raw data must remain untouched so later jobs can replay from the original bytes. Keep this layer focused on faithful storage of the source payload plus ingestion metadata.

Use this folder partitioning convention:

s3://your-bucket/crm/salesforce/accounts/year=2026/month=09/day=20/ s3://your-bucket/crm/hubspot/contacts/year=2026/month=09/day=20/

Hive-style key=value partition directories are recognized automatically by Athena, Redshift Spectrum, Spark, and DuckDB. This pattern enables partition pruning without extra catalog configuration. Plain date folders like 2026-09-20 require manual partition declaration.

Each file in the landing zone should carry three metadata fields alongside the source payload:

  • ingested_at: UTC timestamp of when the batch landed
  • batch_id: a UUID identifying the pipeline run for lineage and replay
  • source_system: for example, salesforce, hubspot, or dynamics365

Store raw files as compressed JSON or the source’s native CSV format. Apply schema-on-read for raw landing zones. Projecting structure during processing instead of enforcing it at landing prevents an unexpected source field from blocking ingestion.

Convert data to Parquet in the silver layer. Converting JSON to Parquet can cut query costs by more than 70% for regular reports.

Write a commit marker file (_manifest.json) last, after all data parts are written. If the manifest exists, the partition is complete and safe to load. If it does not exist, the next sweep will overwrite it.

Step 4: Implement Incremental Loads And SCD Type 2

Incremental loads keep CRM pipelines efficient. Full reloads are expensive and rarely necessary for most CRM objects. Use watermark columns to extract only changed records. When a source table exposes an updated_at or LastModifiedDate column, use it as the incremental watermark with a half-open interval filter.

-- Salesforce incremental extract filter WHERE LastModifiedDate >= '2026-09-19T00:00:00Z' AND LastModifiedDate < '2026-09-20T00:00:00Z'

Dimension tables such as Accounts and Contacts often need full history. Implement SCD Type 2 for these tables. Type 2 Slowly Changing Dimensions create a new row for each change with validity timestamps, which preserves historical analysis while keeping a clear current state.

-- SCD Type 2 MERGE for Accounts (Databricks / Spark SQL) MERGE INTO silver.accounts AS target USING ( SELECT id, name, industry, annual_revenue, last_modified_date, ingested_at, batch_id FROM bronze.accounts_staging ) AS source ON target.id = source.id AND target.is_current = TRUE WHEN MATCHED AND ( target.name <> source.name OR target.industry <> source.industry OR target.annual_revenue <> source.annual_revenue ) THEN UPDATE SET target.valid_to = source.ingested_at, target.is_current = FALSE WHEN NOT MATCHED THEN INSERT ( id, name, industry, annual_revenue, valid_from, valid_to, is_current, batch_id ) VALUES ( source.id, source.name, source.industry, source.annual_revenue, source.ingested_at, NULL, TRUE, source.batch_id ); -- Insert new version for updated rows INSERT INTO silver.accounts SELECT id, name, industry, annual_revenue, ingested_at AS valid_from, NULL AS valid_to, TRUE AS is_current, batch_id FROM bronze.accounts_staging s WHERE EXISTS ( SELECT 1 FROM silver.accounts t WHERE t.id = s.id AND t.is_current = FALSE AND t.valid_to = s.ingested_at );

Databricks’ AUTO CDC feature lets users declaratively define change data capture logic, including keys, sequencing, delete handling, and SCD Type 1 or Type 2 storage, with minimal code. This approach replaces many complex MERGE INTO statements.

Step 5: Orchestrate And Schedule

Orchestration keeps incremental CRM ingestion reliable over time. The following Airflow DAG skeleton handles incremental loads with retry logic:

from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta default_args = { "owner": "data-engineering", "retries": 3, "retry_delay": timedelta(minutes=5), "retry_exponential_backoff": True, } with DAG( dag_id="crm_salesforce_incremental", default_args=default_args, schedule_interval="0 2 * * *", # 02:00 UTC daily start_date=datetime(2026, 1, 1), catchup=True, # enables backfill max_active_runs=1, ) as dag: extract = PythonOperator( task_id="extract_accounts", python_callable=run_bulk_api_extract, op_kwargs={"object": "Account", "watermark_col": "LastModifiedDate"}, ) land = PythonOperator( task_id="land_to_s3", python_callable=write_to_landing_zone, ) merge = PythonOperator( task_id="scd2_merge", python_callable=run_scd2_merge, op_kwargs={"object": "accounts"}, ) extract >> land >> merge

For Azure-based stacks, Azure Data Factory’s Copy Activity with the Dynamics 365 connector handles entity export natively. Azure Data Factory is recommended for connector-based movement and pipeline orchestration into ADLS Gen2. Enable retry policies on each activity and set a pipeline-level timeout so long-running jobs do not consume your entire API quota.

Step 6: Evaluate Build Versus Buy For Connectors

The build-versus-buy decision depends on connector coverage, engineering capacity, and time-to-value expectations.

A 12-month TCO model for a mid-size SaaS team running 5–10 ingestion sources estimates managed platform costs at $25,000–$90,000 fully loaded annually versus $90,000–$220,000+ for a custom pipeline. Ongoing maintenance usually lands around 0.1–0.2 FTE for managed tools and 0.5–1.5 FTE for custom connector upkeep.

Building a custom batch ETL pipeline for 3–5 sources often costs $10,000–$30,000 over 4–8 weeks with $500–$2,000 per month of ongoing maintenance. For 10 or more sources, costs typically rise to $30,000–$60,000 over 8–16 weeks with $1,000–$3,000 per month of maintenance.

Some situations still point to custom builds. These include cases where the primary data source has no connector, the use case needs sub-minute latency, or compliance rules require that raw data always stay inside the company’s own VPC. Outside those constraints, managed connectors usually win on speed and maintenance burden for standard CRM sources such as Salesforce and HubSpot.

Coffee complements whichever path you choose by improving CRM data quality at the point of capture. Coffee’s AI agent keeps records clean inside the CRM, which reduces downstream pipeline complexity and makes both managed and custom connectors easier to operate.

Compare Coffee To Custom Pipelines

Step 7: Implement PII Governance

CRM objects contain direct identifiers such as email, phone, and full name, along with indirect identifiers such as IP address, device ID, and financial fields. GDPR, CCPA, HIPAA, and SOC 2 all require organizations to know where PII lives, who accessed it, and how to delete it on request.

Implement the following controls at the lake layer:

For deep dives on GDPR right-to-erasure implementation in Delta Lake, see Microsoft’s Azure Databricks GDPR guidance. For AWS Lake Formation column and row filtering, see Hidekazu Konishi’s Lake Formation fine-grained access control guide.

Common Mistakes And Troubleshooting

Most CRM ingestion projects run into the same recurring issues:

  • API rate limiting: HubSpot returns HTTP 429 with a Retry-After header when limits are exceeded. Read the header and implement exponential backoff. Avoid fixed sleep intervals.
  • Pagination mishandling: Salesforce Bulk API 2.0 uses a job and result-set workflow with an Sforce-Locator header, not page numbers. HubSpot uses an after cursor token. Dynamics 365 uses @odata.nextLink and does not support $skip.
  • Schema drift: CRM admins add custom fields without notifying data engineering. Use schema-on-read in the raw zone and detect new columns in the silver merge step before they break downstream models.
  • Missing watermark columns: Some custom objects lack LastModifiedDate. Fall back to comparing source and destination rows by primary key hash, or trigger a full reload for that object only.
  • SCD Type 2 merge failures: A MERGE that matches on id AND is_current = TRUE will silently skip rows if is_current was not set correctly on the previous run. Add a row-count reconciliation check after every merge.
  • Orchestration timeouts: Salesforce REST and SOAP API calls time out after 10 minutes. For large Bulk API jobs, poll asynchronously instead of blocking a synchronous task.

Validation And Success Criteria

Healthy pipelines produce correct data, not just error-free runs. Validate on four dimensions:

  • Row counts: Compare source record counts (via CRM reporting API) against landed row counts in the bronze table after each run.
  • Null checks: Assert that primary key columns (Id, AccountId) contain no nulls in the silver layer.
  • Watermark progression: Confirm that the maximum LastModifiedDate in the bronze table advances with each incremental run. A stale watermark indicates a broken extraction filter.
  • SCD Type 2 history correctness: For a known record that changed, verify exactly one row with is_current = TRUE and at least one row with is_current = FALSE and a non-null valid_to.

Coffee strengthens this validation process by providing ground-truth CRM data from real interactions. Coffee captures activity from emails, calendars, and call transcripts automatically, so teams can compare pipeline output against a trusted behavioral record when they debug discrepancies.

Variations And Scaling Considerations

Pipeline design should match team size, connector count, and data volume:

Scale Clean CRM Data With Coffee

Frequently Asked Questions

What Are The Two Main Types Of Data Ingestion?

The two main types are batch ingestion and streaming ingestion. Batch ingestion extracts data on a schedule such as hourly, daily, or weekly, and loads it in discrete chunks. It is simpler to build, easier to reprocess, and sufficient for most CRM analytics use cases where data that is one to several hours old is acceptable. Streaming ingestion captures and delivers data continuously as events occur, which suits fraud detection, real-time dashboards, and operational AI applications that require data within seconds or minutes. For CRM pipelines, most teams start with batch ingestion using incremental watermarks and add streaming CDC only when business requirements clearly demand sub-minute freshness.

Is Data Ingestion The Same As ETL?

Data ingestion and ETL overlap, but they describe different scopes. ETL (Extract, Transform, Load) describes a three-phase process where data is transformed before it reaches the destination. Modern CRM pipelines more commonly follow ELT (Extract, Load, Transform). Raw data lands in the lake first, and transformations run in-warehouse afterward using tools like dbt. This pattern decouples ingestion from analytical logic so teams can revise business rules without re-extracting from the source. Data ingestion refers specifically to the extraction and landing phases, regardless of whether transformation happens before or after landing.

How Do I Export Data From A CRM?

The right export method depends on volume and freshness requirements. For Salesforce, the Bulk API 2.0 is the correct choice for any extraction exceeding a few thousand records. It handles up to 150 million records per 24-hour rolling window and abstracts batch management automatically. For HubSpot, use list endpoints with cursor-based pagination for full object exports, and batch read endpoints (up to 100 object IDs per request) to reduce call volume. Avoid the CRM’s native UI export for pipeline use. HubSpot’s UI export silently omits engagement bodies, attachment binary content, and the full association graph. For Dynamics 365, use the Dataverse Web API with server-driven paging via @odata.nextLink. In all cases, authenticate using OAuth 2.0 or JWT Bearer flows and avoid hard-coded API keys.

How Do I Ingest Dynamics 365 Entities Into A Data Lake With Data Factory?

Azure Data Factory’s native Dynamics 365 (Microsoft Dataverse) connector supports Microsoft Entra service principal authentication (authenticationType “AADServicePrincipal”) and enables incremental extraction by manually filtering on the modifiedon attribute in a FetchXML query, rather than through a built-in watermark mechanism. The general pattern is straightforward. Create a Linked Service pointing to your Dynamics 365 environment with a service principal credential. Create a Dataset referencing the target entity, such as account, contact, or opportunity. Configure a Copy Activity with a source filter on modifiedon >= @pipeline().parameters.watermark_start. Write the output to ADLS Gen2 in Parquet format, partitioned by ingestion date. Store the high-water mark in an Azure SQL table or pipeline parameter and update it at the end of each successful run. For large entities, enable parallel copy and set the degree of copy parallelism to match your Dataverse API throughput. The Dataverse Web API returns up to 5,000 rows per page, and Data Factory’s built-in pagination handles the @odata.nextLink loop automatically.

Conclusion

Building a reliable CRM data ingestion pipeline means navigating API rate limits, pagination patterns, schema drift, and incremental watermarks. Then come the harder problems: SCD Type 2 merge logic and PII governance. All of this work happens before the first analyst query runs. The engineering burden is real and ongoing because APIs change, schemas drift, and watermarks can break silently.

Coffee reduces that burden by improving CRM data quality at the source. Coffee’s AI agent captures CRM data automatically from emails, calendars, and call transcripts so Salesforce or HubSpot hold accurate ground-truth data from the start. Your lake then receives high-quality records without as many manual fixes or fragile extraction scripts.

Use this guide when you need to design and run the pipeline itself. When you want to reduce pipeline complexity by fixing CRM data at the point of capture, Coffee provides a practical path forward.

Start Capturing Clean CRM Data

Read Next