data-catalog-builder

name	data-catalog-builder
description	Data catalog and metadata management covering catalog architecture, DataHub and Amundsen setup, automated metadata ingestion, data lineage tracking, data discovery UX, classification and tagging, data governance integration, business glossaries, ownership models, and search optimization for enterprise data discoverability. Use when the user asks about data catalog builder, data catalog builder best practices, or needs guidance on data catalog builder implementation. Do NOT use when the user needs a different specialized skill or is asking about an unrelated technology domain.
license	Apache-2.0
metadata	{"author":"foundry-skills","version":"1.0.0","tags":"data-science sql architecture","category":"data-engineering","subcategory":"data-modeling","depends":"","disclaimer":"none","difficulty":"intermediate"}

Data Catalog Builder

Overview

A data catalog is the organized inventory of data assets in an organization, making data discoverable, understandable, and trustworthy. This skill covers building and operating data catalogs that serve as the single source of truth for metadata, lineage, ownership, and data governance across the enterprise.

Catalog Architecture

                    +-------------------+
                    |   Search & UI     |
                    |   (Discovery)     |
                    +--------+----------+
                             |
                    +--------v----------+
                    |  Metadata Store   |
                    |  (Graph DB / ES)  |
                    +--------+----------+
                             |
          +------------------+------------------+
          |                  |                  |
  +-------v------+  +-------v------+  +--------v-----+
  | Ingestion    |  | Lineage      |  | Governance   |
  | Connectors   |  | Tracker      |  | Engine       |
  +-------+------+  +-------+------+  +--------+-----+
          |                  |                  |
  +-------v------+  +-------v------+  +--------v-----+
  | Data Sources |  | Orchestrators|  | Policies     |
  | (DBs, Lakes, |  | (Airflow,    |  | (Access,     |
  |  APIs, Files)|  |  Spark, dbt) |  |  PII, Class.)|
  +--------------+  +--------------+  +--------------+

DataHub Implementation

Configuration and Ingestion

# datahub-ingestion-recipe.yaml
source:
  type: postgres
  config:
    host_port: "warehouse.company.com:5432"
    database: analytics
    username: "${DATAHUB_PG_USER}"
    password: "${DATAHUB_PG_PASSWORD}"
    include_tables: true
    include_views: true
    schema_pattern:
      allow:
        - "analytics.*"
        - "staging.*"
      deny:
        - ".*_tmp$"
        - ".*_backup$"
    profiling:
      enabled: true
      profile_table_level_only: false
      max_number_of_fields_to_profile: 50
      profile_table_row_count_estimate_only: false
    classification:
      enabled: true
      classifiers:
        - type: datahub
          config:
            pii_detection: true
            info_types_config:
              - EMAIL_ADDRESS
              - PHONE_NUMBER
               
               


   
  
     
     

Programmatic Metadata Emission

from datahub.emitter.mce_builder import make_dataset_urn
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import (
    DatasetPropertiesClass,
    OwnershipClass,
    OwnerClass,
    OwnershipTypeClass,
    TagAssociationClass,
    GlossaryTermAssociationClass,
    InstitutionalMemoryClass,
    InstitutionalMemoryMetadataClass,
)
from datahub.metadata.com.linkedin.pegasus2avro.mxe import MetadataChangeEvent

emitter = DatahubRestEmitter("[reference URL]")

dataset_urn = make_dataset_urn(platform="snowflake", name="analytics.orders")

# Set dataset properties
properties = DatasetPropertiesClass(
    name="Orders",
    description="Customer orders with line items and payment details",
    customProperties={
        "data_tier": "gold",
        "refresh_frequency": "hourly",
        "sla_freshness": "2 hours",
        "row_count_estimate": "50M",
        "pii_classification": "contains_pii",
    },
    qualifiedName="snowflake.prod.analytics.orders",
)

# Set ownership
ownership = OwnershipClass(
    owners=[
        OwnerClass(
            owner="urn:li:corpuser:[e-mail]",
            type=OwnershipTypeClass.DATA_STEWARD,
        ),
        OwnerClass(
            owner="urn:li:corpGroup:commerce-team",
            type=OwnershipTypeClass.TECHNICAL_OWNER,
        ),
    ]
)

# Emit metadata
emitter.emit_mce(MetadataChangeEvent(
    proposedSnapshot=DatasetSnapshotClass(
        urn=dataset_urn,
        aspects=[properties, ownership],
    )
))

Lineage Emission

from datahub.emitter.mce_builder import make_dataset_urn
from datahub.metadata.schema_classes import (
    UpstreamClass,
    UpstreamLineageClass,
    DatasetLineageTypeClass,
)

# Define lineage: orders_analytics depends on raw_orders + customers
lineage = UpstreamLineageClass(
    upstreams=[
        UpstreamClass(
            dataset=make_dataset_urn("postgres", "raw.orders"),
            type=DatasetLineageTypeClass.TRANSFORMED,
        ),
        UpstreamClass(
            dataset=make_dataset_urn("postgres", "raw.customers"),
            type=DatasetLineageTypeClass.TRANSFORMED,
        ),
    ]
)

emitter.emit_mce(MetadataChangeEvent(
    proposedSnapshot=DatasetSnapshotClass(
        urn=make_dataset_urn("snowflake", "analytics.order_metrics"),
        aspects=[lineage],
    )
))

Data Lineage Tracking

Automated Lineage from SQL Parsing

from sqllineage.runner import LineageRunner

def extract_lineage_from_sql(sql: str) -> dict:
    """Parse SQL to extract table-level lineage."""
    runner = LineageRunner(sql)

    return {
        "source_tables": [str(t) for t in runner.source_tables()],
        "target_tables": [str(t) for t in runner.target_tables()],
        "columns": {
            str(col): [str(src) for src in runner.get_column_lineage()]
            for col in runner.target_tables()
        }
    }

# Example
sql = """
INSERT INTO analytics.order_summary
SELECT
    o.region,
    c.segment,
    SUM(o.amount) as total_revenue,
    COUNT(DISTINCT o.customer_id) as customer_count
FROM raw.orders o
JOIN raw.customers c ON o.customer_id = c.id
WHERE o.status = 'completed'
GROUP BY o.region, c.segment
"""

lineage = extract_lineage_from_sql(sql)
# {'source_tables': ['raw.orders', 'raw.customers'],
#  'target_tables': ['analytics.order_summary']}

dbt Lineage Integration

# dbt_project integration with DataHub
# profiles.yml includes DataHub emission

# models/marts/order_metrics.sql
# Lineage is automatically captured from ref() and source()
{{
  config(
    materialized='incremental',
    unique_key='order_date || region',
    meta={
      'owner': 'analytics-team',
      'tier': 'gold',
      'contains_pii': false,
    }
  )
}}

SELECT
    DATE_TRUNC('day', o.created_at) as order_date,
    o.region,
    COUNT(*) as order_count,
    SUM(o.total_amount) as revenue
FROM {{ ref('stg_orders') }} o
WHERE o.status != 'cancelled'
{% if is_incremental() %}
  AND o.created_at > (SELECT MAX(order_date) FROM {{ this }})
{% endif %}
GROUP BY  

Business Glossary

Glossary Structure

# business-glossary.yaml
glossary:
  - term: "Customer"
    definition: >
      An individual or organization that has completed at least one
      purchase. Prospects and leads are not customers until first purchase.
    owner: commerce-team
    related_terms: [Account, Buyer, Client]
    authoritative_source: "analytics.dim_customers"
    aliases: [buyer, purchaser, client]

  - term: "Revenue"
    definition: >
      Net amount received from completed orders after discounts, returns,
      and refunds. Excludes taxes, shipping, and pending orders.
    owner: finance-team
    calculation: "SUM(order_total) - SUM(refunds) - SUM(discounts)"
    related_terms: [GMV, Net Revenue, ARR]
    authoritative_source: "analytics.fact_revenue"
    important_notes:
      - "Differs from GMV which includes cancelled orders"
      - "Does not include subscription revenue (see ARR)"

  - term: "Active User"
    definition: >
      A user who has performed at least one qualifying action within
      the measurement period. Qualifying actions: login, purchase,
      API call, or dashboard view. Bot traffic is excluded.
     
    
       
       
       
     

Data Classification and Tagging

Classification Framework

class DataClassifier:
    """Automated data classification for catalog enrichment."""

    CLASSIFICATION_RULES = {
        'pii_email': {
            'pattern': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$',
            'tag': 'pii:email',
            'governance': 'restricted',
            'match_threshold': 0.7,
        },
        'pii_phone': {
            'pattern': r'^\+?[\d\s\-\(\)]{7,15}$',
            'tag': 'pii:phone',
            'governance': 'restricted',
            'match_threshold': 0.6,
        },
        'pii_ssn': {
            'pattern': r'^\d{3}-\d{2}-\d{4}$',
            'tag': 'pii:ssn',
            'governance': 'highly_restricted',
            'match_threshold': 0.5,
        },
        'financial': {
            'column_names': ['revenue', 'amount', 'price', 'cost', 'salary'],
            'tag': 'data_class:financial',
            'governance': 'confidential',
        },
    }

    def classify_column(self, column_name, sample_values):
        tags = []
        for rule_name, rule  .CLASSIFICATION_RULES.items():
               rule:
                match_rate = ._pattern_match_rate(sample_values, rule[])
                 match_rate >= rule.get(, ):
                    tags.append(rule[])
               rule:
                 (kw  column_name.lower()  kw  rule[]):
                    tags.append(rule[])
         tags

     ():
        classification = {}
         col  df.columns:
            sample = df[col].dropna().head()
            classification[col] = {
                : .classify_column(col, sample),
                : ._highest_governance(col, sample),
            }
         classification

Ownership Model

ownership_model:
  roles:
    data_owner:
      description: "Business stakeholder accountable for data quality and access"
      responsibilities:
        - Approve access requests
        - Define data quality requirements
        - Approve schema changes
        - Ensure regulatory compliance

    technical_owner:
      description: "Engineering team maintaining the pipeline and infrastructure"
      responsibilities:
        - Maintain ingestion pipelines
        - Fix data quality issues
        - Manage schema evolution
        - Monitor SLAs

    data_steward:
      description: "Subject matter expert who curates metadata and definitions"
      responsibilities:
        - Maintain descriptions and documentation
            
            
            

  
     
     
     
     

Search Optimization

Improving Data Discovery

class CatalogSearchOptimizer:
    """Strategies for improving catalog search quality."""

    def build_search_document(self, dataset_metadata):
        """Create enriched search document for indexing."""
        return {
            "name": dataset_metadata.name,
            "qualified_name": dataset_metadata.qualified_name,
            "description": dataset_metadata.description,
            "columns": [c.name for c in dataset_metadata.columns],
            "column_descriptions": [c.description for c in dataset_metadata.columns],
            "tags": dataset_metadata.tags,
            "glossary_terms": dataset_metadata.glossary_terms,
            "owner": dataset_metadata.owner,
            "team": dataset_metadata.team,
            "tier": dataset_metadata.tier,
            "usage_count": dataset_metadata.query_count_30d,
            "freshness": dataset_metadata.last_updated,
            # Boost fields for relevance
            "aliases": dataset_metadata.aliases,
            "upstream_names": [u.name for u in dataset_metadata.upstream],
            "downstream_count": len(dataset_metadata.downstream),
        }

    def rank_results(self, results, user_context):
        """Re-rank search results based on user context."""
        for result in results:
            score = result.base_score
            # Boost datasets owned by user's team
             result.team == user_context.team:
                score *= 
            
            score *=  + (result.usage_count / )
            
             result.tier == :
                score *= 
            
             result.days_since_update > :
                score *= 
            result.final_score = score
         (results, key= r: r.final_score, reverse=)

Catalog Implementation Checklist

Foundation:
  [ ] Select catalog platform (DataHub, Amundsen, OpenMetadata, or custom)
  [ ] Deploy infrastructure (graph DB, search index, API)
  [ ] Configure authentication and authorization
  [ ] Set up automated ingestion for top 10 data sources

Metadata Quality:
  [ ] Define business glossary with 20+ core terms
  [ ] Assign ownership to all critical datasets
  [ ] Write descriptions for top 50 most-queried datasets
  [ ] Classify PII and sensitive columns
  [ ] Tag datasets with data tier (bronze/silver/gold)

Lineage:
  [ ] Integrate SQL parsing for automated column lineage
  [ ] Connect dbt for model lineage
  [ ] Connect orchestrator (Airflow/Dagster) for pipeline lineage
  [ ] Verify end-to-end lineage for 5 critical data paths

Governance:
  [ ] Define data classification levels
  [ ] Implement access request workflow
  [ ] Configure PII detection rules
  [ ] Set up ownership review cadence
  [ ] Define deprecation process for retired datasets

Adoption:
  [ ] Integrate catalog link into BI tool descriptions
  [ ] Add catalog search to Slack bot
  [ ] Track catalog usage metrics (searches, page views)
  [ ] Run onboarding sessions for data consumers
  [ ] Measure discovery time before/after catalog

When to Use

Use this skill when:

• Designing or implementing data catalog builder solutions
• Reviewing or improving existing data catalog builder approaches
• Making architectural or implementation decisions about data catalog builder
• Learning data catalog builder patterns and best practices
• Troubleshooting data catalog builder-related issues

Do NOT use this skill when:

• The question is about a fundamentally different technology domain
• A more specific sibling skill covers the exact topic needed
• The user needs a complete hands-on tutorial rather than expert guidance

Output Format

# Data Catalog Builder Analysis

## Context Assessment
[Situation summary and constraints]

## Recommended Approach
[Primary recommendation with rationale]

## Implementation Steps
1. [Step with specific details]
2. [Step with specific details]
3. [Step with specific details]

## Trade-offs and Considerations
- [Key trade-off 1]
- [Key trade-off 2]

## Next Steps
- [Immediate action item]
- [Follow-up action item]

Example

Input: "Help me implement data catalog builder for a medium-scale production application"

Output: A structured analysis covering current state assessment, recommended data catalog builder approach with specific patterns, implementation roadmap with milestones, and risk mitigation strategies tailored to the application scale and constraints.

Edge Cases

• Legacy system integration: When data catalog builder must coexist with legacy approaches, provide a gradual migration path rather than a complete rewrite
• Scale mismatch: When the solution complexity exceeds the project scale, recommend a simpler approach and note when to revisit
• Team skill gaps: When the team lacks experience with the recommended approach, include learning resources and simpler alternatives
• Conflicting requirements: When constraints conflict (e.g., performance vs. maintainability), explicitly state the trade-off and recommend based on stated priorities