GraphQL Mutations for Bulk Operations

Performing operations on many entities through the UI is time-consuming. Users need efficient ways to bulk update tags, owners, domains, and other metadata across multiple datasets.

You Might Be Asking:

• How do I bulk add tags to multiple datasets?
• Can I update metadata for many entities at once?
• What's the most efficient way to do bulk operations?

1. Bulk add tags to multiple datasets:

mutation batchAddTags {
  batchAddTags(
    input: {
      tagUrns: [
        "urn:li:tag:PII",
        "urn:li:tag:Sensitive"
      ]
      resources: [
        {
          resourceUrn: "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.table1,PROD)"
        },
        {
          resourceUrn: "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.table2,PROD)"
        },
        {
          resourceUrn: "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.table3,PROD)"
        }
      ]
    }
  )
}

2. Bulk assign domains:

mutation batchSetDomain {
  batchSetDomain(
    input: {
      domainUrn: "urn:li:domain:Finance"
      resources: [
        {
          resourceUrn: "urn:li:dataset:(urn:li:dataPlatform:snowflake,finance.sales,PROD)"
        },
        {
          resourceUrn: "urn:li:dataset:(urn:li:dataPlatform:snowflake,finance.revenue,PROD)"
        }
      ]
    }
  )
}

3. Bulk update owners using Python SDK:

from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.emitter.mce_builder import make_dataset_urn, make_user_urn
from datahub.metadata.schema_classes import OwnershipClass, OwnerClass, OwnershipTypeClass

emitter = DatahubRestEmitter("http://localhost:8080", token="YOUR_TOKEN")

# List of datasets to update
datasets = [
    "db.schema.table1",
    "db.schema.table2",
    "db.schema.table3"
]

owner_to_add = make_user_urn("john.doe")

for dataset_name in datasets:
    dataset_urn = make_dataset_urn("snowflake", dataset_name)

    ownership = OwnershipClass(
        owners=[
            OwnerClass(
                owner=owner_to_add,
                type=OwnershipTypeClass.TECHNICAL_OWNER
            )
        ]
    )

    emitter.emit_mcp(
        entity_urn=dataset_urn,
        aspect_name="ownership",
        aspect=ownership
    )

    print(f"Updated ownership for {dataset_name}")

4. Bulk operations with search results:

import requests

# GraphQL client
def graphql_query(query, variables=None):
    response = requests.post(
        "http://localhost:8080/api/graphql",
        json={"query": query, "variables": variables},
        headers={"Authorization": f"Bearer {TOKEN}"}
    )
    return response.json()

# Search for datasets and bulk tag them
search_query = """
query searchDatasets($query: String!, $start: Int!, $count: Int!) {
  search(input: {
    type: DATASET
    query: $query
    start: $start
    count: $count
    filters: [{field: "platform", values: ["snowflake"]}]
  }) {
    searchResults {
      entity {
        urn
      }
    }
  }
}
"""

# Get all datasets matching criteria
all_urns = []
start = 0
count = 100

while True:
    result = graphql_query(search_query, {
        "query": "finance",
        "start": start,
        "count": count
    })

    results = result['data']['search']['searchResults']
    if not results:
        break

    for r in results:
        all_urns.append(r['entity']['urn'])

    start += count

# Bulk add tags to all found datasets
batch_size = 50
for i in range(0, len(all_urns), batch_size):
    batch = all_urns[i:i+batch_size]

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

    graphql_query(mutation, {
        "tagUrns": ["urn:li:tag:Finance"],
        "resources": [{"resourceUrn": urn} for urn in batch]
    })

    print(f"Tagged batch {i//batch_size + 1}")

5. Bulk remove tags:

mutation batchRemoveTags {
  batchRemoveTags(
    input: {
      tagUrns: ["urn:li:tag:Deprecated"]
      resources: [
        {resourceUrn: "urn:li:dataset:(...)"},
        {resourceUrn: "urn:li:dataset:(...)"}
      ]
    }
  )
}

6. Bulk update custom properties:

# Bulk update custom properties across datasets
from datahub.metadata.schema_classes import DatasetPropertiesClass

datasets_to_update = {
    "db.schema.table1": {"team": "analytics", "tier": "gold"},
    "db.schema.table2": {"team": "analytics", "tier": "silver"},
    "db.schema.table3": {"team": "finance", "tier": "gold"}
}

for dataset_name, properties in datasets_to_update.items():
    dataset_urn = make_dataset_urn("snowflake", dataset_name)

    dataset_props = DatasetPropertiesClass(
        customProperties=properties
    )

    emitter.emit_mcp(
        entity_urn=dataset_urn,
        aspect_name="datasetProperties",
        aspect=dataset_props
    )

Batch mutations have size limits (typically 50-100 entities per request). For very large bulk operations, implement batching and error handling. Consider rate limiting to avoid overwhelming the system. Always test on a small subset first.

• GraphQL API
• Python SDK