GraphQL Pagination with scrollAcrossEntities

When retrieving large datasets from DataHub via GraphQL, users need efficient pagination mechanisms to fetch all entities without overwhelming the API or running into memory constraints. The standard `search` query can become slow and inefficient for large result sets, especially when trying to retrieve all datasets with specific metadata fields. The `scrollAcrossEntities` query provides a more efficient scroll-based pagination approach specifically designed for bulk data extraction, but many users are unfamiliar with its syntax and usage patterns.

You Might Be Asking:

• How do I fetch all datasets from DataHub efficiently?
• What's the difference between search and scrollAcrossEntities?
• How do I implement pagination in GraphQL?
• How do I minimize data transfer when querying thousands of entities?
• Can I filter fields to reduce response size?
• What's the maximum number of results per page?

Understanding Pagination Methods

DataHub offers two main pagination approaches:

1. search - Offset-based pagination, better for UI/user-facing queries
2. scrollAcrossEntities - Scroll-based pagination, better for bulk exports

Using scrollAcrossEntities

This is the recommended approach for bulk data extraction:

query scrollDatasets {
  scrollAcrossEntities(
    input: {
      types: [DATASET]
      count: 1000
      # scrollId: null for first page, then use returned scrollId
      query: "*"
      # Optional filters
      filters: []
    }
  ) {
    searchResults {
      entity {
        ... on Dataset {
          urn
          name
          properties {
            name
            description
            customProperties {
              key
              value
            }
          }
          schemaMetadata {
            fields {
              fieldPath
              type
              nativeDataType
              description
            }
          }
        }
      }
    }
    nextScrollId  # Use this for next page
  }
}

Complete Python Example with Pagination

import requests
import json
from typing import List, Dict, Any, Optional

class DataHubScrollAPI:
    def __init__(self, gms_host: str, token: str):
        self.gms_host = gms_host
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
        self.graphql_url = f"{gms_host}/api/graphql"

    def scroll_all_entities(
        self,
        entity_type: str = "DATASET",
        page_size: int = 1000,
        query: str = "*",
        filters: Optional[List[Dict]] = None
    ) -> List[Dict[str, Any]]:
        """
        Scroll through all entities using pagination
        """
        all_entities = []
        scroll_id = None

        while True:
            # Build GraphQL query
            graphql_query = """
                query scrollEntities($input: ScrollAcrossEntitiesInput!) {
                    scrollAcrossEntities(input: $input) {
                        searchResults {
                            entity {
                                urn
                                type
                                ... on Dataset {
                                    name
                                    platform {
                                        name
                                    }
                                    properties {
                                        name
                                        description
                                    }
                                    schemaMetadata {
                                        fields {
                                            fieldPath
                                            type
                                            nativeDataType
                                        }
                                    }
                                }
                            }
                        }
                        nextScrollId
                    }
                }
            """

            variables = {
                "input": {
                    "types": [entity_type],
                    "count": page_size,
                    "query": query,
                    "scrollId": scroll_id,
                    "filters": filters or []
                }
            }

            # Execute query
            response = requests.post(
                self.graphql_url,
                json={"query": graphql_query, "variables": variables},
                headers=self.headers
            )

            if response.status_code != 200:
                raise Exception(f"GraphQL query failed: {response.text}")

            data = response.json()

            if "errors" in data:
                raise Exception(f"GraphQL errors: {data['errors']}")

            results = data["data"]["scrollAcrossEntities"]["searchResults"]
            scroll_id = data["data"]["scrollAcrossEntities"]["nextScrollId"]

            # Add results to collection
            all_entities.extend([r["entity"] for r in results])

            print(f"Fetched {len(results)} entities. Total: {len(all_entities)}")

            # Check if we're done
            if not scroll_id or len(results) == 0:
                break

        return all_entities

# Usage
api = DataHubScrollAPI(
    gms_host="http://localhost:8080",
    token="YOUR_ACCESS_TOKEN"
)

# Fetch all datasets
all_datasets = api.scroll_all_entities(
    entity_type="DATASET",
    page_size=1000
)

print(f"Total datasets fetched: {len(all_datasets)}")

# Save to file
with open("all_datasets.json", "w") as f:
    json.dump(all_datasets, f, indent=2)

Optimized Query for Minimal Data Transfer

Request only the fields you need:

query minimalScroll {
  scrollAcrossEntities(
    input: {
      types: [DATASET]
      count: 1000
      scrollId: $scrollId
    }
  ) {
    searchResults {
      entity {
        urn  # Just URN
        ... on Dataset {
          name  # And name
          # Omit large fields like schema, properties if not needed
        }
      }
    }
    nextScrollId
  }
}

Filtering Results

Apply filters to narrow down results:

Example for Snowflake datasets with specific tags. This can be adapted for other platforms and filter criteria.

query filteredScroll {
  scrollAcrossEntities(
    input: {
      types: [DATASET]
      count: 1000
      query: "*"
      filters: [
        {
          field: "platform"
          values: ["snowflake"]
        },
        {
          field: "tags"
          values: ["urn:li:tag:PII"]
        },
        {
          field: "domains"
          values: ["urn:li:domain:Finance"]
        }
      ]
    }
  ) {
    searchResults {
      entity {
        urn
      }
    }
    nextScrollId
  }
}

Multi-Entity Type Query

Fetch multiple entity types in one scroll:

query multiTypeScroll {
  scrollAcrossEntities(
    input: {
      types: [DATASET, DASHBOARD, CHART]
      count: 1000
      scrollId: $scrollId
    }
  ) {
    searchResults {
      entity {
        urn
        type
        ... on Dataset {
          properties {
            name
          }
        }
        ... on Dashboard {
          properties {
            name
          }
        }
        ... on Chart {
          properties {
            name
          }
        }
      }
    }
    nextScrollId
  }
}

Error Handling and Retry Logic

import time
from typing import Optional

def scroll_with_retry(
    api: DataHubScrollAPI,
    max_retries: int = 3,
    backoff_factor: int = 2
) -> List[Dict]:
    """Scroll with exponential backoff retry"""

    all_entities = []
    scroll_id: Optional[str] = None
    retry_count = 0

    while True:
        try:
            # Make API call
            response = api.scroll_page(scroll_id)

            # Reset retry count on success
            retry_count = 0

            # Process results
            entities = response.get("searchResults", [])
            all_entities.extend(entities)

            # Get next scroll ID
            scroll_id = response.get("nextScrollId")

            if not scroll_id or len(entities) == 0:
                break

        except Exception as e:
            retry_count += 1

            if retry_count >= max_retries:
                print(f"Max retries reached. Stopping at {len(all_entities)} entities")
                break

            wait_time = backoff_factor ** retry_count
            print(f"Error: {e}. Retrying in {wait_time}s...")
            time.sleep(wait_time)

    return all_entities

Comparison: search vs scrollAcrossEntities

# search - Offset-based, good for UI
query searchDatasets {
  search(
    input: {
      type: DATASET
      query: "*"
      start: 0      # Offset
      count: 10     # Page size
    }
  ) {
    total          # Total count available
    searchResults {
      entity { urn }
    }
  }
}

# scrollAcrossEntities - Cursor-based, good for exports
query scrollDatasets {
  scrollAcrossEntities(
    input: {
      types: [DATASET]
      count: 1000   # Larger pages supported
      scrollId: null # Cursor
    }
  ) {
    searchResults {
      entity { urn }
    }
    nextScrollId   # Next cursor
  }
}

Performance Optimization Tips

1. Use appropriate page size:

# Too small = many API calls
count: 100  # Avoid for large datasets

# Optimal range
count: 1000  # Good balance

# Too large = memory issues
count: 10000  # May cause timeouts

2. Request only needed fields:

# Bad - requests everything
... on Dataset {
  properties {
    # All fields
  }
  schemaMetadata {
    # All schema
  }
  ownership {
    # All owners
  }
}

# Good - selective fields
... on Dataset {
  urn
  name
  # Only what you need
}

3. Use filters to reduce result set:

# Filter at query time, not in application
filters: [
  { field: "platform", values: ["snowflake"] }
]

Export All Datasets Script

Complete production-ready script:

#!/usr/bin/env python3
"""
Export all DataHub datasets to JSON
"""
import argparse
import json
from dataclasses import dataclass
from typing import List
import requests

@dataclass
class ExportConfig:
    datahub_url: str
    token: str
    entity_type: str
    output_file: str
    page_size: int = 1000

def export_entities(config: ExportConfig):
    """Export all entities to JSON file"""

    url = f"{config.datahub_url}/api/graphql"
    headers = {
        "Authorization": f"Bearer {config.token}",
        "Content-Type": "application/json"
    }

    query = """
        query scroll($input: ScrollAcrossEntitiesInput!) {
            scrollAcrossEntities(input: $input) {
                searchResults {
                    entity {
                        urn
                        type
                    }
                }
                nextScrollId
            }
        }
    """

    all_entities = []
    scroll_id = None
    page = 0

    while True:
        page += 1
        print(f"Fetching page {page}...")

        variables = {
            "input": {
                "types": [config.entity_type],
                "count": config.page_size,
                "scrollId": scroll_id,
                "query": "*"
            }
        }

        response = requests.post(
            url,
            json={"query": query, "variables": variables},
            headers=headers,
            timeout=30
        )

        response.raise_for_status()
        data = response.json()

        results = data["data"]["scrollAcrossEntities"]["searchResults"]
        all_entities.extend([r["entity"] for r in results])

        scroll_id = data["data"]["scrollAcrossEntities"]["nextScrollId"]

        print(f"  Fetched {len(results)} entities (total: {len(all_entities)})")

        if not scroll_id or len(results) == 0:
            break

    # Save results
    with open(config.output_file, "w") as f:
        json.dump(all_entities, f, indent=2)

    print(f"\nExported {len(all_entities)} entities to {config.output_file}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--url", required=True)
    parser.add_argument("--token", required=True)
    parser.add_argument("--type", default="DATASET")
    parser.add_argument("--output", default="export.json")

    args = parser.parse_args()

    config = ExportConfig(
        datahub_url=args.url,
        token=args.token,
        entity_type=args.type,
        output_file=args.output
    )

    export_entities(config)

• Scroll sessions expire after 5 minutes of inactivity
• Maximum page size is typically 10,000 but 1,000 is recommended
• scrollAcrossEntities is more efficient than offset-based pagination for large result sets
• Results are eventually consistent - new entities during scroll may be missed
• For real-time data, use search with filters instead
• Consider implementing progress tracking for long-running exports
• Rate limiting may apply - check with your DataHub admin
• This is the preferred method for data migration and analysis tasks

• GraphQL API
• API Overview

5029, 5224