Skip to main content

Semantic Models & Metrics

Why Would You Use Semantic Models and Metrics?​

Semantic Models and Metrics let you describe a logical layer over your physical data: a semanticModel groups one or more logical datasets (each a "Semantic Model Dataset" subtype), exposes dimensions and measures via schema-field-anchored semanticFieldAnnotations, and serves as the backing model for metric entities. Together they form the lineage chain Metric → SemanticModel → Logical Dataset → Physical Dataset, giving consumers a stable, source-agnostic surface for analytics, governance, and AI-assisted exploration.

This mirrors modeling that already exists inside the Snowflake connector, lifted into the high-level SDK so any producer (connector or direct SDK user) can emit the same entities without re-implementing the aspect wiring.

Goal Of This Guide​

This guide will show you how to:

  • Build a SemanticModel with two logical datasets, schema fields, and a relationship.
  • Emit metric entities backed by the model, including one metric derived from another.
  • Serialize every emitted MCP to a file and inspect the resulting aspect shapes.

Prerequisites​

For this tutorial, you need DataHub SDK v2 (datahub.sdk.*) installed. If you are running the example from the metadata-ingestion package, the venv is set up by ../gradlew :metadata-ingestion:installDev.

Build a Semantic Model with Logical Datasets and Metrics​

The example below builds the full lineage chain Metric -> SemanticModel -> Logical Dataset -> Physical Dataset using the high-level datahub.sdk builders, then writes every emitted MCP to a JSON file so the resulting aspect shapes can be inspected.

# Inlined from /metadata-ingestion/examples/library/semantic_model_create.py
"""Emit a semantic model with two logical datasets and two metrics.

This example builds the full lineage chain
``Metric -> SemanticModel -> Logical Dataset -> Physical Dataset`` using the
high-level ``datahub.sdk`` builders, then writes every emitted MCP to a JSON
file so the resulting aspect shapes can be inspected.

Run with::

    python -m examples.library.semantic_model_create

The output is written to ``semantic_model_create.json`` in the current
directory. Inspect it to confirm the aspect shapes match the producer contract:
URN patterns, the ``Semantic Model Dataset`` subtype, the
``semanticModelProperties`` back-refs, the schemaField-anchored
``semanticFieldAnnotation`` MCPs, the required-expression fallback
(``ORDERS.order_id``), aiContext-only-when-non-empty, and the absence of
``metricUpstreams`` for semantic-model-backed metrics.
"""

import json
from typing import Any, List

from datahub.emitter.mce_builder import make_dataset_urn
from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.metadata.schema_classes import (
    DialectClass,
    ERModelRelationshipCardinalityClass,
    SemanticFieldTypeClass,
)
from datahub.metadata.urns import SemanticModelUrn
from datahub.sdk import (
    AiContextInput,
    DialectExpressionInput,
    Metric,
    SemanticFieldInput,
    SemanticModel,
    SemanticModelDataset,
    SemanticModelRelationshipInput,
)
from datahub.sdk.entity import Entity


def build_graph() -> tuple[SemanticModel, List[Entity]]:
    platform = "snowflake"
    model_urn = SemanticModelUrn(platform=platform, path="analytics", id="orders_model")

    orders_ds = SemanticModelDataset(
        platform=platform,
        name="analytics.orders_model.orders_ds",
        semantic_model=model_urn,
        alias="ORDERS",
        schema=[
            SemanticFieldInput(
                field_path="order_id",
                type="int",
                semantic_type=SemanticFieldTypeClass.DIMENSION,
                is_part_of_key=True,
            ),
            # Foreign key the ORDERS -> CUSTOMERS relationship joins on.
            SemanticFieldInput(
                field_path="customer_id",
                type="int",
                semantic_type=SemanticFieldTypeClass.DIMENSION,
            ),
            SemanticFieldInput(
                field_path="order_ts",
                type="timestamp",
                semantic_type=SemanticFieldTypeClass.DIMENSION,
                is_time_dimension=True,
            ),
            SemanticFieldInput(
                field_path="amount",
                type="float",
                semantic_type=SemanticFieldTypeClass.MEASURE,
                expression=DialectExpressionInput(
                    expression="SUM(amount)", dialect=DialectClass.SNOWFLAKE
                ),
                aggregation_function="SUM",
                ai_context=AiContextInput(synonyms=["revenue"]),
            ),
        ],
        upstreams=[make_dataset_urn(platform, "raw.orders")],
    )

    customers_ds = SemanticModelDataset(
        platform=platform,
        name="analytics.orders_model.customers_ds",
        semantic_model=model_urn,
        alias="CUSTOMERS",
        schema=[
            SemanticFieldInput(
                field_path="customer_id",
                type="int",
                semantic_type=SemanticFieldTypeClass.DIMENSION,
                is_part_of_key=True,
            ),
            SemanticFieldInput(
                field_path="customer_name",
                type="varchar",
                semantic_type=SemanticFieldTypeClass.DIMENSION,
            ),
        ],
        upstreams=[make_dataset_urn(platform, "raw.customers")],
    )

    model = SemanticModel(
        platform=platform,
        path="analytics",
        id="orders_model",
        name="Orders Model",
        description="A semantic model over the raw orders and customers tables.",
        datasets=[orders_ds, customers_ds],
        relationships=[
            SemanticModelRelationshipInput(
                from_alias="ORDERS",
                from_columns=["customer_id"],
                to_alias="CUSTOMERS",
                to_columns=["customer_id"],
                name="orders_to_customers",
                cardinality=ERModelRelationshipCardinalityClass.N_ONE,
            )
        ],
        ai_context=AiContextInput(
            synonyms=["orders model"],
            instructions="Use for revenue and customer analytics.",
        ),
    )

    total_revenue = Metric(
        platform=platform,
        path="analytics",
        id="total_revenue",
        semantic_model=str(model.urn),
        name="Total Revenue",
        description="Sum of all order amounts.",
        expression=DialectExpressionInput(
            expression="SUM(ORDERS.amount)", dialect=DialectClass.SNOWFLAKE
        ),
        ai_context=AiContextInput(synonyms=["revenue"]),
    )
    double_revenue = Metric(
        platform=platform,
        path="analytics",
        id="double_revenue",
        semantic_model=str(model.urn),
        name="Double Revenue",
        expression="2 * total_revenue",
        derived_from=[total_revenue.urn],
    )

    return model, [orders_ds, customers_ds, total_revenue, double_revenue]


def main() -> None:
    model, entities = build_graph()

    all_mcps: list[MetadataChangeProposalWrapper] = []
    all_mcps.extend(model.as_mcps())
    for entity in entities:
        all_mcps.extend(entity.as_mcps())

    records: list[dict[str, Any]] = [dict(mcp.to_obj()) for mcp in all_mcps]
    with open("semantic_model_create.json", "w") as f:
        json.dump(records, f, indent=2, default=str)
    print(f"Wrote {len(all_mcps)} MCPs to semantic_model_create.json")

    # When emitting to a live server instead of a file, call the opt-in
    # preflight helper first to get a clear error on an unsupported server:
    #
    #   from datahub.sdk import DataHubClient, require_metrics_support
    #   client = DataHubClient(server=..., token=...)
    #   require_metrics_support(client)  # raises if the server version is too old
    #   for entity in [model, *entities]:
    #       client.entities.upsert(entity)


if __name__ == "__main__":
    main()

What the SDK emits for you​

When you call entity.as_mcps() on each builder, the SDK produces the full aspect set and wires the lineage chain automatically:

  • semanticModel: a Status, a SemanticModelInfo (with datasets preserving insertion order, plus optional relationships), and a model-level AiContext only when non-empty.
  • Logical datasets: each gets SubTypes([SEMANTIC_MODEL_DATASET]), a SemanticModelProperties(alias, semanticModel=<model urn>) back-ref, a SchemaMetadata with the declared fields, and — when upstreams is provided — an UpstreamLineage to the physical datasets. For every field, the SDK emits a schemaField-anchored semanticFieldAnnotation (with expression auto-synthesized as f"{alias}.{field_path}" when not provided) and, when non-empty, a field-anchored aiContext.
  • metrics: each gets Status, MetricInfo (with semanticModel=<model urn> back-ref and an optional expression; the expression is never fabricated when omitted), MetricRelationships (always emitted, even with empty derivedFrom, so hasParentMetric indexes as false), and an AiContext only when non-empty.

Note that the SDK does not populate metricUpstreams for semantic-model-backed metrics — the lineage chain is expressed entirely through metricInfo.semanticModel, semanticModelInfo.datasets, and the logical dataset's own upstreamLineage.

Expected Output​

Running the example writes semantic_model_create.json in the working directory. Open it and verify the aspect shapes match the producer contract:

  • URN patterns: urn:li:semanticModel:(urn:li:dataPlatform:snowflake,analytics,orders_model), urn:li:metric:(urn:li:dataPlatform:snowflake,analytics,total_revenue), and urn:li:dataset:(urn:li:dataPlatform:snowflake,analytics.orders_model.orders_ds,PROD).
  • Logical datasets carry the Semantic Model Dataset subtype.
  • Each logical dataset's semanticModelProperties points back at the model URN with the right alias.
  • semanticFieldAnnotation MCPs are anchored on schemaField URNs and the expression falls back to ORDERS.order_id when not explicitly provided.
  • aiContext is only present on fields/entities that had non-empty inputs.
  • No metricUpstreams aspect is emitted for the metrics.

API Reference​

For the full surface area of each builder, see the SDK Entities Reference.

  • SemanticModel — datahub.sdk.semantic_model.SemanticModel
  • SemanticModelDataset — datahub.sdk.semantic_model.SemanticModelDataset
  • Metric — datahub.sdk.metric.Metric

Server compatibility​

The semanticModel, metric, and logical-dataset entities require a server build that registers the semantic-model metadata model. Emitting to a server that does not register these aspects fails loudly — the server rejects the unregistered aspect and emit_mcps raises.

For a clear, actionable error instead of a server-side rejection, call the opt-in preflight helper before emitting:

from datahub.sdk import DataHubClient, require_metrics_support

client = DataHubClient(server="...", token="...")
require_metrics_support(client)  # raises if the server version is too old

The helper delegates to RestServiceConfig.supports_feature: it raises when the server reports a version that does not support these entities, and fails open when there is no version signal to check (the operator is then responsible for running a build that includes the model). It is not wired into DataHubClient.upsert automatically — call it explicitly when you want the preflight.

Read-modify-write caveat for logical datasets​

Per-field semanticFieldAnnotation and field-level aiContext on a SemanticModelDataset are create-only. A logical dataset shares the dataset entity type, so client.entities.get(<dataset urn>) hydrates it as a base Dataset — the field-anchored annotations live on schemaField URNs, not in the dataset's aspect bag, and are not carried back on a read. To update a logical dataset, rebuild a fresh SemanticModelDataset and re-attach its fields via the schema constructor kwarg rather than read-modify-writing the fetched Dataset.