REST API Rate Limiting and Throttling

High-volume API usage is being rate limited or throttled, causing ingestion failures or application errors. Understanding rate limits and implementing proper backoff strategies is important for reliable integrations.

Common Error Messages:

• 429 Too Many Requests
• Rate limit exceeded
• Quota exceeded
• Throttling applied

You Might Be Asking:

• What are the DataHub API rate limits?
• How do I handle rate limiting?
• Can I increase rate limits?

1. Implement exponential backoff:

import time
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

def create_session_with_retries():
    """
    Create session with automatic retry and backoff
    """
    session = requests.Session()

    retry_strategy = Retry(
        total=5,  # Total number of retries
        backoff_factor=2,  # Exponential backoff: 2, 4, 8, 16, 32 seconds
        status_forcelist=[429, 500, 502, 503, 504],  # Retry on these status codes
        method_whitelist=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE", "POST"]
    )

    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)

    return session

# Usage
session = create_session_with_retries()
response = session.post(
    "http://localhost:8080/api/graphql",
    json={"query": "..."},
    headers={"Authorization": f"Bearer {token}"}
)

2. Implement rate limiting in client:

from ratelimit import limits, sleep_and_retry
import time

class RateLimitedDataHubClient:
    """
    DataHub client with built-in rate limiting
    """

    # Configure rate limits (adjust based on your needs)
    CALLS_PER_MINUTE = 100
    CALLS_PER_SECOND = 10

    def __init__(self, gms_url, token):
        self.gms_url = gms_url
        self.token = token
        self.session = create_session_with_retries()

    @sleep_and_retry
    @limits(calls=CALLS_PER_SECOND, period=1)
    @limits(calls=CALLS_PER_MINUTE, period=60)
    def make_request(self, query, variables=None):
        """
        Make GraphQL request with rate limiting
        """
        response = self.session.post(
            f"{self.gms_url}/api/graphql",
            json={"query": query, "variables": variables},
            headers={"Authorization": f"Bearer {self.token}"}
        )

        if response.status_code == 429:
            # Handle rate limit with additional backoff
            retry_after = int(response.headers.get('Retry-After', 60))
            print(f"Rate limited. Waiting {retry_after} seconds...")
            time.sleep(retry_after)
            return self.make_request(query, variables)

        response.raise_for_status()
        return response.json()

# Usage
client = RateLimitedDataHubClient("http://localhost:8080", "token")
result = client.make_request("{ me { username } }")

3. Batch operations to reduce API calls:

# Instead of individual calls
for dataset in datasets:
    add_tags(dataset, ["PII"])  # 100 API calls

# Batch into single call
batch_add_tags(datasets, ["PII"])  # 1 API call

# Batch implementation
def batch_add_tags(dataset_urns, tag_urns, batch_size=50):
    """
    Add tags to multiple datasets in batches
    """
    for i in range(0, len(dataset_urns), batch_size):
        batch = dataset_urns[i:i+batch_size]

        mutation = """
        mutation batchAddTags($tagUrns: [String!]!, $resources: [ResourceRefInput!]!) {
          batchAddTags(input: {
            tagUrns: $tagUrns
            resources: $resources
          })
        }
        """

        client.make_request(mutation, {
            "tagUrns": tag_urns,
            "resources": [{"resourceUrn": urn} for urn in batch]
        })
# Small delay between batches
        time.sleep(0.1)

4. Use connection pooling for REST emitter:

from datahub.emitter.rest_emitter import DatahubRestEmitter
from requests.adapters import HTTPAdapter

# Configure emitter with connection pooling
emitter = DatahubRestEmitter(
    gms_server="http://localhost:8080",
    token="your-token",
    timeout_sec=60,
    retry_status_codes=[429, 500, 502, 503],
    retry_max_times=5
)

# Configure session
adapter = HTTPAdapter(
    pool_connections=10,
    pool_maxsize=50,
    max_retries=5
)
emitter._session.mount("http://", adapter)
emitter._session.mount("https://", adapter)

5. Monitor API usage:

import logging
from datetime import datetime, timedelta

class APIUsageTracker:
    """
    Track API usage and warn about approaching limits
    """

    def __init__(self, warning_threshold=0.8):
        self.calls = []
        self.warning_threshold = warning_threshold
        self.max_calls_per_minute = 100

    def record_call(self):
        """Record an API call"""
        self.calls.append(datetime.now())
        self._cleanup_old_calls()
        self._check_threshold()

    def _cleanup_old_calls(self):
        """Remove calls older than 1 minute"""
        cutoff = datetime.now() - timedelta(minutes=1)
        self.calls = [call for call in self.calls if call > cutoff]

    def _check_threshold(self):
        """Warn if approaching rate limit"""
        current_rate = len(self.calls)
        threshold = self.max_calls_per_minute * self.warning_threshold

        if current_rate >= threshold:
            logging.warning(
                f"API usage at {current_rate}/{self.max_calls_per_minute} calls/min. "
                f"Approaching rate limit!"
            )

    def get_current_rate(self):
        """Get current calls per minute"""
        self._cleanup_old_calls()
        return len(self.calls)

# Usage
tracker = APIUsageTracker()

for dataset in datasets:
    tracker.record_call()
    emit_metadata(dataset)

    # Adaptive throttling
    if tracker.get_current_rate() > 80:
        time.sleep(1)  # Slow down

6. Configure server-side rate limits:

# For self-hosted DataHub, configure in application.yml
datahub-gms:
  env:
    # API rate limiting configuration
    - name: RATE_LIMIT_ENABLED
      value: "true"

    - name: RATE_LIMIT_REQUESTS_PER_MINUTE
      value: "1000"

    - name: RATE_LIMIT_REQUESTS_PER_HOUR
      value: "10000"

    # Per-user limits
    - name: RATE_LIMIT_PER_USER_ENABLED
      value: "true"

    - name: RATE_LIMIT_PER_USER_REQUESTS_PER_MINUTE
      value: "100"

Rate limits protect DataHub infrastructure from overload. Implement client-side throttling before hitting server limits. Use batch operations whenever possible. For high-volume scenarios, consider Kafka emitter instead of REST. Monitor 429 responses and adjust accordingly.

• REST API Guide
• Python SDK