Data Cataloging & Metadata Management â Tools, Lineage & Discovery
In this tutorial, you'll learn about Data Cataloging & Metadata Management. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Data cataloging is the practice of creating a centralized inventory of data assets â tables, columns, dashboards, pipelines, and metrics â with metadata, lineage, documentation, and discovery capabilities so users can find, understand, and trust data.
What You'll Learn
By the end of this tutorial, you'll understand the components of a data catalog, metadata harvesting strategies, column-level lineage tracking, business glossary management, catalog tools (Datahub, Amundsen, Atlan), and how to maintain catalog quality at scale.
Why It Matters
In organizations with hundreds of tables and thousands of dashboards, data discovery becomes the bottleneck. Analysts spend 40-60% of their time finding and understanding data rather than analyzing it. Without a catalog, teams maintain tribal knowledge, build duplicate datasets, and make decisions based on misunderstood data. DodaTech's data catalog serves 200+ data assets across engineering, product, and marketing teams, reducing data discovery time from hours to seconds.
Real-World Use
LinkedIn built Datahub (now open-source) to catalog 100,000+ datasets. Airbnb uses Amundsen for data discovery across 10,000+ tables. Uber's Databook catalogs petabytes of data with automated lineage from source to dashboard.
Data Catalog Architecture
flowchart TB
subgraph "Metadata Sources"
A[Databases] --> M[Metadata Ingestion]
B[Data Warehouses] --> M
C[BI Tools] --> M
D[Pipeline Orchestrators] --> M
E[Streaming Systems] --> M
end
subgraph "Catalog Platform"
M --> F[(Metadata Store)]
F --> G[Search Index]
F --> H[Lineage Graph]
F --> I[Business Glossary]
end
subgraph "Consumers"
G --> J[Data Discovery UI]
H --> K[Impact Analysis]
I --> L[Governance Reports]
end
style M fill:#f90,color:#fff
style F fill:#f90,color:#fff
Prerequisites: Understanding of SQL and data warehousing concepts. Familiarity with Python helps for metadata API integrations.
What Is a Data Catalog?
Think of a data catalog like a library catalog for your data warehouse. Without it, finding the right table is like walking into a library where books have no labels, no sections, and no index. The catalog makes every dataset discoverable, understandable, and trustable.
Core Components
| Component | Purpose | Example |
|---|---|---|
| Metadata Store | Central database of asset information | Table schemas, column types, row counts |
| Search Index | Full-text search across all assets | Find "customer revenue" across tables |
| Lineage Graph | Track data flow from source to consumption | Table A feeds dashboard B |
| Business Glossary | Standardized business definitions | "ARR" = Annual Recurring Revenue |
| Profiling | Column statistics for quality signals | Null %, distinct count, distribution |
| Ownership | Who maintains each dataset | Team contacts, documentation owners |
Automated Metadata Harvesting
# metadata_harvester.py
# Extract metadata from databases and publish to catalog
import json
from datetime import datetime
class MetadataHarvester:
def __init__(self, catalog_url=None):
self.catalog_url = catalog_url
self.assets = []
def extract_schema_from_postgres(self, host, database, schema):
"""Simulate extracting table and column metadata from PostgreSQL."""
print(f"[HARVEST] Extracting schema from {host}/{database}/{schema}")
tables = {
"orders": {
"columns": [
{"name": "order_id", "type": "INTEGER", "nullable": False, "description": "Unique order identifier"},
{"name": "customer_id", "type": "VARCHAR(50)", "nullable": False, "description": "Foreign key to customers"},
{"name": "amount", "type": "DECIMAL(12,2)", "nullable": False, "description": "Order total amount"},
{"name": "status", "type": "VARCHAR(20)", "nullable": True, "description": "Order status code"},
{"name": "created_at", "type": "TIMESTAMP", "nullable": False, "description": "Order creation time"},
],
"row_count": 1500000,
"size_mb": 450,
},
"customers": {
"columns": [
{"name": "customer_id", "type": "VARCHAR(50)", "nullable": False, "description": "Unique customer ID"},
{"name": "email", "type": "VARCHAR(255)", "nullable": False, "description": "Customer email address"},
{"name": "signup_date", "type": "DATE", "nullable": True, "description": "When customer signed up"},
],
"row_count": 250000,
"size_mb": 85,
},
}
for table_name, meta in tables.items():
asset = {
"type": "table",
"name": f"{database}.{schema}.{table_name}",
"schema": database,
"table": table_name,
"columns": meta["columns"],
"statistics": {
"row_count": meta["row_count"],
"size_mb": meta["size_mb"],
},
"ingested_at": datetime.now().isoformat(),
}
self.assets.append(asset)
print(f" Found table: {asset['name']} ({meta['row_count']} rows, {meta['size_mb']}MB)")
return self.assets
def extract_lineage_from_airflow(self):
"""Simulate extracting pipeline lineage."""
print(f"\n[LINEAGE] Extracting pipeline dependencies")
lineage = [
{"source": "raw_db.public.orders", "target": "analytics.stg_orders", "pipeline": "orders_etl", "type": "ETL"},
{"source": "raw_db.public.customers", "target": "analytics.stg_customers", "pipeline": "customers_etl", "type": "ETL"},
{"source": "analytics.stg_orders", "target": "analytics.fct_orders", "pipeline": "dbt_run", "type": "dbt"},
{"source": "analytics.stg_customers", "target": "analytics.dim_customers", "pipeline": "dbt_run", "type": "dbt"},
{"source": "analytics.fct_orders", "target": "bi.dashboard.daily_sales", "pipeline": "looker_sync", "type": "BI"},
]
for entry in lineage:
print(f" {entry['source']} -> {entry['target']} ({entry['type']})")
return lineage
def profile_columns(self):
"""Simulate column profiling."""
print(f"\n[PROFILE] Profiling columns...")
profiles = {
"analytics.orders.status": {
"null_percentage": 2.5,
"unique_values": 5,
"frequent_values": {"completed": 45.2, "pending": 22.1, "shipped": 18.5, "cancelled": 10.0, "refunded": 4.2},
"type": "VARCHAR",
},
"analytics.orders.amount": {
"null_percentage": 0.0,
"min": 0.01,
"max": 50000.0,
"mean": 275.45,
"std_dev": 890.23,
},
}
for col, profile in profiles.items():
print(f" {col}: {profile.get('null_percentage', 0)}% null, "
f"{profile.get('unique_values', 'N/A')} unique values")
return profiles
harvester = MetadataHarvester()
assets = harvester.extract_schema_from_postgres("prod-db", "ecommerce", "public")
lineage = harvester.extract_lineage_from_airflow()
profiles = harvester.profile_columns()
print(f"\nSummary: {len(assets)} assets harvested, {len(lineage)} lineage entries, {len(profiles)} profiles")
Expected output:
[HARVEST] Extracting schema from prod-db/ecommerce/public
Found table: ecommerce.public.orders (1500000 rows, 450MB)
Found table: ecommerce.public.customers (250000 rows, 85MB)
[LINEAGE] Extracting pipeline dependencies
raw_db.public.orders -> analytics.stg_orders (ETL)
raw_db.public.customers -> analytics.stg_customers (ETL)
analytics.stg_orders -> analytics.fct_orders (dbt)
analytics.stg_customers -> analytics.dim_customers (dbt)
analytics.fct_orders -> bi.dashboard.daily_sales (BI)
[PROFILE] Profiling columns...
analytics.orders.status: 2.5% null, 5 unique values
analytics.orders.amount: 0.0% null, N/A unique values
Summary: 2 assets harvested, 5 lineage entries, 2 profiles
Column-Level Lineage
Lineage answers critical questions: "Where does this column come from?" and "What breaks if I change this table?"
# column_lineage.py
# Track column-level lineage across pipeline stages
class ColumnLineage:
def __init__(self):
self.nodes = {}
self.edges = []
def add_column(self, table, column, data_type, description=""):
node_id = f"{table}.{column}"
self.nodes[node_id] = {
"table": table,
"column": column,
"type": data_type,
"description": description,
}
return node_id
def add_transformation(self, source_table, source_column, target_table, target_column, logic):
source_id = f"{source_table}.{source_column}"
target_id = f"{target_table}.{target_column}"
self.edges.append({
"source": source_id,
"target": target_id,
"logic": logic,
})
def trace_upstream(self, table, column):
"""Find all upstream sources for a given column."""
target = f"{table}.{column}"
visited = set()
def trace(node_id, depth=0):
if node_id in visited:
return
visited.add(node_id)
indent = " " * depth
node = self.nodes.get(node_id, {})
print(f"{indent}{node.get('column', node_id)} ({node.get('type', '?')})")
for edge in self.edges:
if edge["target"] == node_id:
print(f"{indent} <- [{edge['logic']}]")
trace(edge["source"], depth + 1)
print(f"Lineage for {table}.{column}:")
trace(target)
def impact_analysis(self, table, column):
"""Find all downstream dependencies for a given column."""
source = f"{table}.{column}"
visited = set()
def trace(node_id, depth=0):
if node_id in visited:
return
visited.add(node_id)
node = self.nodes.get(node_id, {})
indent = " " * depth
print(f"{indent}{node.get('column', node_id)} ({node.get('type', '?')})")
for edge in self.edges:
if edge["source"] == node_id:
print(f"{indent} -> [{edge['logic']}]")
trace(edge["target"], depth + 1)
print(f"Impact analysis for {table}.{column}:")
trace(source)
lineage = ColumnLineage()
lineage.add_column("raw.orders", "order_total", "DECIMAL", "Raw order amount from source")
lineage.add_column("raw.orders", "tax_amount", "DECIMAL", "Tax calculated at checkout")
lineage.add_column("stg.orders", "net_amount", "DECIMAL", "Order total minus tax")
lineage.add_column("analytics.fct_orders", "revenue", "DECIMAL", "Final revenue amount")
lineage.add_column("bi.daily_sales", "total_revenue", "DECIMAL", "Aggregated daily revenue")
lineage.add_transformation("raw.orders", "order_total", "stg.orders", "net_amount", "order_total - tax_amount")
lineage.add_transformation("stg.orders", "net_amount", "analytics.fct_orders", "revenue", "direct mapping")
lineage.add_transformation("analytics.fct_orders", "revenue", "bi.daily_sales", "total_revenue", "SUM(revenue) GROUP BY date")
lineage.trace_upstream("bi.daily_sales", "total_revenue")
print()
lineage.impact_analysis("raw.orders", "order_total")
Expected output:
Lineage for bi.daily_sales.total_revenue:
total_revenue (DECIMAL)
<- [SUM(revenue) GROUP BY date]
revenue (DECIMAL)
<- [direct mapping]
net_amount (DECIMAL)
<- [order_total - tax_amount]
order_total (DECIMAL)
tax_amount (DECIMAL)
Impact analysis for raw.orders.order_total:
order_total (DECIMAL)
-> [order_total - tax_amount]
net_amount (DECIMAL)
-> [direct mapping]
revenue (DECIMAL)
-> [SUM(revenue) GROUP BY date]
total_revenue (DECIMAL)
Business Glossary
A business glossary defines metrics and dimensions in business terms, independent of technical implementation:
# business_glossary.py
class BusinessGlossary:
def __init__(self):
self.terms = {}
def add_term(self, name, definition, formula, examples=None, synonyms=None, domain=None):
self.terms[name] = {
"name": name,
"definition": definition,
"formula": formula,
"examples": examples or [],
"synonyms": synonyms or [],
"domain": domain or "General",
"created_at": datetime.now().isoformat(),
}
def search(self, query):
query = query.lower()
results = []
for name, term in self.terms.items():
if query in name.lower() or query in term["definition"].lower():
results.append(term)
return results
def link_to_column(self, term_name, table, column):
if term_name in self.terms:
if "mapped_columns" not in self.terms[term_name]:
self.terms[term_name]["mapped_columns"] = []
self.terms[term_name]["mapped_columns"].append({"table": table, "column": column})
def glossary_report(self):
print(f"\n=== Business Glossary ===")
print(f"{'Term':<20} {'Domain':<20} {'Definition'}")
print("-" * 80)
for name, term in sorted(self.terms.items()):
cols = ", ".join(f"{c['table']}.{c['column']}" for c in term.get("mapped_columns", []))
print(f"{name:<20} {term['domain']:<20} {term['definition'][:60]}...")
if cols:
print(f"{'':>20} {'':<20} Mapped to: {cols}")
print(f"\nTotal terms: {len(self.terms)}")
from datetime import datetime
glossary = BusinessGlossary()
glossary.add_term("Monthly Recurring Revenue", "Subscription revenue normalized to a monthly value",
"SUM(subscription_price) WHERE status = 'active'", domain="Finance")
glossary.add_term("Customer Lifetime Value", "Total revenue expected from a customer over their lifetime",
"AVG(order_value) * AVG(purchase_frequency) * AVG(customer_lifetime_months)", domain="Marketing")
glossary.add_term("Churn Rate", "Percentage of customers who cancel in a given period",
"cancelled_customers / total_customers * 100", domain="Product")
glossary.link_to_column("Monthly Recurring Revenue", "analytics.fct_subscriptions", "mrr")
glossary.link_to_column("Customer Lifetime Value", "analytics.dim_customers", "ltv")
glossary.glossary_report()
Expected output:
=== Business Glossary ===
Term Domain Definition
--------------------------------------------------------------------------------
Churn Rate Product Percentage of customers who cancel in a given...
Customer Lifetime Value Marketing Total revenue expected from a customer over t...
Monthly Recurring Reven Finance Subscription revenue normalized to a monthly ...
... ... Mapped to: analytics.fct_subscriptions.mrr
Total terms: 3
Common Data Cataloging Mistakes
1. Building a Catalog Without Automation
Manual metadata entry is abandoned within weeks. Every table, column, and lineage must be harvested automatically via API. If it's not automated, it's not a catalog â it's a wiki.
2. No Business Context
Technical metadata (column type, nullable) without business context (what does this column mean, who owns it) is useless to analysts. Always pair schema metadata with descriptions, tags, and ownership.
3. Ignoring Data Consumers
A catalog built for data engineers only includes technical metadata. Add consumer-focused features: certified datasets, popularity scores, sample queries, and dashboard lineage.
4. Lineage Only at Table Level
Table-level lineage tells you that Table A feeds Dashboard B. Column-level lineage tells you exactly which column in Table A maps to which field in Dashboard B. Always harvest column-level lineage.
5. No Catalog Governance
Without ownership, stale metadata accumulates. Set policies: every table must have an owner, descriptions must be updated quarterly, deprecated assets must be tagged. Enforce via automated scans.
Practice Questions
1. What is the difference between technical metadata and business metadata? Technical metadata describes the data structure (column names, types, constraints, row counts). Business metadata describes meaning (definitions, formulas, use cases, data owners, certifications). Both are needed for a useful catalog.
2. How does column-level lineage differ from table-level lineage? Table-level lineage shows data flows between tables: Table A -> Table B. Column-level lineage shows exactly which columns map: Table A.amount -> Table B.revenue. Column-level lineage enables precise impact analysis ("changing this column breaks these 5 dashboards").
3. What is the purpose of a business glossary in a data catalog? A business glossary standardizes metric definitions so everyone uses the same formula for "Monthly Recurring Revenue." It links business terms to physical columns, resolves synonyms, and ensures consistent reporting across teams.
Frequently Asked Questions
{{< faq question="Should I build or buy a data catalog?">}} Buy unless you have a dedicated team for catalog development. Open-source options like Datahub (LinkedIn) and Amundsen (Lyft) are production-proven with active communities. Commercial options like Atlan, Alation, and Collibra offer enterprise features (governance workflows, SSO, role-based access). Building from scratch requires maintaining metadata ingestion, search indexing, lineage Parsing, and a UI â easily 2-3 engineering years. {{< /faq >}}
{{< faq question="How often should metadata be refreshed?">}} Schema metadata: every 6-12 hours (overnight is sufficient). Lineage: after each pipeline run (or daily batch). Profiling: weekly for large tables, daily for critical tables. Freshness: real-time for operational pipelines. Configure refresh cadence per asset type rather than a one-size-fits-all schedule. {{< /faq >}}
Mini Project: Simple Catalog Search
# catalog_search.py
class DataCatalog:
def __init__(self):
self.assets = []
self.index = {}
def add_table(self, database, schema, table, columns, description, tags=None):
asset = {
"type": "table",
"id": f"{database}.{schema}.{table}",
"database": database,
"schema": schema,
"table": table,
"columns": [{"name": c["name"], "type": c["type"], "desc": c.get("desc", "")} for c in columns],
"description": description,
"tags": tags or [],
}
self.assets.append(asset)
words = f"{table} {description} {' '.join(c['name'] for c in columns)} {' '.join(tags or [])}".lower()
for word in words.split():
if word not in self.index:
self.index[word] = []
self.index[word].append(asset["id"])
return asset
def search(self, query):
query_words = query.lower().split()
results = {}
for word in query_words:
if word in self.index:
for asset_id in self.index[word]:
results[asset_id] = results.get(asset_id, 0) + 1
sorted_results = sorted(results.items(), key=lambda x: -x[1])
print(f"\n=== Search: '{query}' ===\n")
if not sorted_results:
print(" No results found.")
return []
for asset_id, score in sorted_results:
asset = next(a for a in self.assets if a["id"] == asset_id)
print(f" {asset['id']} (score: {score})")
print(f" {asset['description']}")
print(f" Columns: {', '.join(c['name'] for c in asset['columns'])}")
print(f" Tags: {', '.join(asset['tags'])}")
print()
return sorted_results
catalog = DataCatalog()
catalog.add_table("prod", "analytics", "daily_revenue",
[{"name": "date", "type": "DATE"}, {"name": "total_revenue", "type": "DECIMAL"},
{"name": "order_count", "type": "INTEGER"}],
"Daily aggregated revenue and order counts", tags=["finance", "kpi"])
catalog.add_table("prod", "analytics", "customer_orders",
[{"name": "customer_id", "type": "VARCHAR"}, {"name": "order_id", "type": "VARCHAR"},
{"name": "amount", "type": "DECIMAL"}, {"name": "order_date", "type": "DATE"}],
"Individual customer order records", tags=["sales", "customers"])
catalog.search("revenue customers")
Expected output:
=== Search: 'revenue customers' ===
prod.analytics.daily_revenue (score: 1)
Daily aggregated revenue and order counts
Columns: date, total_revenue, order_count
Tags: finance, kpi
prod.analytics.customer_orders (score: 1)
Individual customer order records
Columns: customer_id, order_id, amount, order_date
Tags: sales, customers
Related Concepts
What's Next
You now understand data cataloging, metadata management, and lineage tracking. Next, explore data governance frameworks for policy enforcement and Compliance, and learn how Python integrates with catalog APIs for custom metadata ingestion.
- Practice daily â Identify 5 tables in your warehouse and write descriptions for each column
- Build a project â Deploy Datahub or Amundsen locally and connect it to a sample database
- Explore related topics â Check out data contracts, schema registries, and active metadata platforms
Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro