Managing DataProcessInstance Accumulation and Garbage Collection in DataHub – DataHub Customer Support Portal

Area: Observability Issues

Sub-Area: Graph Cluster Size / DataProcessInstance Lifecycle Management

Issue

DataHub environments with high-frequency ingestion pipelines can accumulate large numbers of DataProcessInstance (DPI) entities over time. These entities are created automatically as execution records for each ingestion run and are connected to the broader lineage graph. Because DPIs accumulate multiplicatively with every ingestion run, they can cause specific graph clusters to grow dramatically larger than others — sometimes reaching millions of nodes — which can degrade lineage query performance and graph traversal operations. Users may not immediately realize that DPIs are the root cause of an oversized cluster because DPI entities are typically invisible during standard lineage queries.

Error Messages

Largest WCC: 4,000,000+ nodes (reported by graph analysis tools)
Lineage queries returning unexpectedly high total counts via searchAcrossLineage
Slow or degraded UI performance when rendering lineage graphs for highly connected datasets

You Might Be Asking

Why is one lineage cluster orders of magnitude larger than all others?
What are DataProcessInstance entities and why are there so many of them?
How do I identify what entity types are dominating a large graph cluster?
Does DataHub have a built-in way to expire or garbage collect old DataProcessInstance records?
Will deleting millions of DPI entities impact Kafka or ingestion pipelines?

Solution

Step 1: Analyze Graph Cluster Composition

Use the following Python script with NetworkX and DuckDB to analyze the size and composition of weakly connected components (WCCs) in your lineage graph. Export your graph data to Parquet format from your DataHub backing store before running this script.

import pandas as pd
import networkx as nx
import duckdb
from collections import Counter

# Load and filter edges using DuckDB
edges = duckdb.sql("""
    SELECT source_urn, destination_urn
    FROM parquet_scan('<path-to-your-export>/data.parquet')
    WHERE relationship_type IN ('Consumes', 'UpstreamOf', 'DownstreamOf', 'Produces', 'SiblingOf')
""").df()

# Build a directed graph
G = nx.from_pandas_edgelist(
    edges,
    'source_urn',
    'destination_urn',
    create_using=nx.DiGraph()
)

# Basic graph statistics
print(f"Total edges: {G.number_of_edges()}")
print(f"Total nodes: {G.number_of_nodes()}")
print(f"Largest fan-out: {max(dict(G.out_degree()).values())}")

# Analyze weakly connected components (WCCs)
wcc_sizes = [len(c) for c in nx.weakly_connected_components(G)]
print(f"\nLargest WCC: {max(wcc_sizes):,} nodes")
print(f"Top 5 WCC sizes: {sorted(wcc_sizes, reverse=True)[:5]}")

# Inspect the largest WCC
largest_wcc = max(nx.weakly_connected_components(G), key=len)
wcc_graph = G.subgraph(largest_wcc).copy()
print(f"Edges in largest WCC: {wcc_graph.number_of_edges():,}")

# Find the highest-degree node in the largest WCC
degree_dict = dict(wcc_graph.degree())
largest_node = max(degree_dict, key=degree_dict.get)
print(f"\nLargest connected node: {largest_node}")
print(f"  Total degree: {degree_dict[largest_node]}")
print(f"  In-degree: {wcc_graph.in_degree(largest_node)}")
print(f"  Out-degree: {wcc_graph.out_degree(largest_node)}")

# Break down entity types within the largest WCC
def extract_entity_type(urn):
    try:
        return urn.split(':')[2]
    except Exception:
        return 'unknown'

entity_types = Counter(extract_entity_type(n) for n in largest_wcc)
print(f"\nEntity types in largest WCC: {dict(entity_types)}")

A typical output where DPIs are the dominant driver of cluster size will look like:

Total edges: 14,145,311
Total nodes: 5,421,614
Largest fan-out: 1,339

Largest WCC: 4,025,684 nodes
Top 5 WCC sizes: [4025684, 87090, 58844, 24005, 20627]
Edges in largest WCC: 12,626,516

Entity types in WCC #1:
  dataProcessInstance: 3,422,992   <-- dominant type
  schemaField:           950,331
  dataset:               213,479
  dataJob:                86,229
  chart:                  79,476

If dataProcessInstance entities make up the majority of nodes in your largest cluster, DPI accumulation is the root cause.

Step 2: Confirm via searchAcrossLineage (Optional)

You can cross-check the logical entity count (excluding DPIs) using the GraphQL API:

query {
  searchAcrossLineage(
    input: {
      urn: "urn:li:dataset:(urn:li:dataPlatform:<platform>,<your-dataset-fqn>,PROD)"
      direction: DOWNSTREAM
      start: 0
      orFilters: [
        {
          and: [
            {
              field: "degree"
              condition: EQUAL
              values: ["1", "2", "3+"]
            }
          ]
        }
      ]
    }
  ) {
    isPartial
    total
  }
}

If the logical entity count (e.g., ~75,000) is far smaller than the WCC size (e.g., 4 million), the gap is explained by accumulated DPI entities.

Step 3: Configure DataHub Garbage Collection for DataProcessInstances

DataHub includes a built-in garbage collection source (datahub-gc) that can automatically expire old DataProcessInstance records. Add the following recipe to a scheduled ingestion source in your DataHub deployment:

source:
  type: datahub-gc
  config:
    cleanup_expired_tokens: false
    dataprocess_cleanup:
      enabled: true
      retention_days: 90          # Retain DPIs from the last 90 days (adjust as needed)
      keep_last_n: 10             # Always keep the last N runs regardless of age
      delete_empty_data_jobs: true
      delete_empty_data_flows: true
      batch_size: 500             # Keep batches small to limit Kafka pressure
      delay: 0.5                  # Seconds between delete batches

sink:
  type: datahub-rest
  config:
    server: "https://<your-instance>.datahubproject.io"
    token: "<your-datahub-api-token>"

Key configuration parameters to tune:

retention_days: Number of days of DPI history to retain. Use 90 days if you have monthly recurring pipeline runs you wish to preserve.
keep_last_n: Guarantees the most recent N runs are never deleted, regardless of age. Recommended minimum: 10.
batch_size: Controls how many deletions are sent per Kafka batch. Reduce this value (e.g., to 100–250) when performing initial cleanup of very large DPI backlogs.
delay: Pause in seconds between batches. Increase this value to reduce Kafka topic lag during large initial cleanups.

Step 4: Perform a Phased Initial Cleanup for Large Backlogs

If your instance has millions of accumulated DPI entities, do not attempt a single bulk deletion. Follow this phased approach to avoid overwhelming Kafka and impacting ingestion:

Start with a small test batch. Set batch_size: 100 and run the garbage collector once. Monitor Kafka consumer lag before proceeding.
Observe lag metrics. Check your Kafka monitoring dashboard (or DataHub's built-in metrics) to confirm lag is recovering between batches. If lag spikes significantly, reduce batch_size further or increase delay.
Determine a safe daily throughput. Based on observed lag behavior, establish a maximum safe number of deletions per day for your environment.
Run the garbage collector on a schedule. Once the initial backlog is cleared, schedule the GC ingestion to run daily or weekly to prevent future accumulation. The keep_last_n and retention_days parameters will prevent the cluster from growing unbounded.
Avoid running large batch deletes concurrently with peak ingestion windows to minimize disruption to normal pipeline operations.

Additional Notes

Why DPIs cause multiplicative growth: Each time an ingestion pipeline runs, DataHub creates a new DataProcessInstance entity linked to the associated DataJob and upstream/downstream datasets. For pipelines that run frequently (e.g., hourly or daily) against datasets with broad lineage, the number of DPI nodes connected to the lineage cluster compounds with every run. Over months or years, this can cause a single dataset's connected component to grow far larger than its logical lineage footprint would suggest.

DPIs are not returned by standard lineage queries: The searchAcrossLineage API filters to logical entities (datasets, dashboards, etc.) by default. DPIs generally do not appear in the DataHub UI's lineage visualization. As a result, the cluster growth is invisible to end users until it is measured directly at the graph storage layer.

Kafka impact during deletion: Deleting large numbers of entities in DataHub emits tombstone events to Kafka. Sending too many delete events in rapid succession can cause consumer lag to spike, which queues normal ingestion events behind deletions and may temporarily degrade ingestion throughput or UI responsiveness. Always throttle large initial cleanup operations using batch_size and delay.

Retention policy guidance: For pipelines that run on monthly schedules, configure retention_days to at least 90 days to ensure historical run records are not prematurely deleted. For pipelines that run hourly or daily, shorter retention windows (7–30 days) combined with keep_last_n: 10 are generally sufficient.

Proactive monitoring recommendation: If you observe lineage query performance degradation, graph WCC size analysis (as shown in Step 1 above) should be a first-line diagnostic step. Scheduling regular garbage collection ingestion from the outset — rather than reactively after accumulation — is strongly recommended for any DataHub deployment with high-frequency pipelines.