New Dagster Openmetadata Ingestion Pipeline

pyproject.toml

@@ -22,13 +22,13 @@ dependencies = [
     "dagster-slack ~=0.26.1",
     "dagster-webserver ~= 1.10",
     "dbt-duckdb ~= 1.9.0",
-    "dbt-trino ~= 1.9.0",
     "flatten-dict ~= 0.4.2",
     "fsspec ~=2025.2.0",
     "gcsfs (>=2025.2.0,<2025.3.0)",
     "httpx ~= 0.28.0",
     "hvac ~= 2.3.0",
     "jsonlines ~= 4.0.0",
+    "openmetadata-ingestion[trino]>=1.6.4.0",
     "polars ~= 1.19",
     "pyarrow ~=19.0.0",
     "pydantic ~= 2.10.0",

src/ol_orchestrate/assets/platform/openmetadata.py

@@ -0,0 +1,31 @@
+from dagster import (
+    AssetExecutionContext,
+    AssetKey,
+    DataVersion,
+    Output,
+    asset,
+)
+from metadata.workflow.metadata import MetadataWorkflow
+
+from ol_orchestrate.partitions.openmetadata import TRINO_DAILY_PARTITIONS
+from ol_orchestrate.resources.openmetadata.trino import trino_config
+
+
+@asset(
+    description=("An instance of metadata from our Trino database."),
+    group_name="platform",
+    key=AssetKey(["platform", "database", "trino", "metadata"]),
+    partitions_def=TRINO_DAILY_PARTITIONS,
+)
+def trino_metadata(context: AssetExecutionContext):
+    date = context.partition_key
+    workflow = MetadataWorkflow.create(trino_config)
+    workflow.execute()
+    workflow.raise_from_status()
+    workflow.print_status()
+    workflow.stop()
+    return Output(
+        workflow,
+        data_version=DataVersion(workflow["published_version"]),
+        metadata={"source": "Trino", "date": date},
+    )

src/ol_orchestrate/definitions/platform/openmetadata.py

@@ -0,0 +1,30 @@
+from dagster import create_repository_using_definitions_args
+from dagster_aws.s3 import S3Resource
+
+from ol_orchestrate.assets.platform.openmetadata import trino_metadata
+from ol_orchestrate.lib.constants import DAGSTER_ENV, VAULT_ADDRESS
+from ol_orchestrate.resources.secrets.vault import Vault
+
+if DAGSTER_ENV == "dev":
+    vault = Vault(vault_addr=VAULT_ADDRESS, vault_auth_type="github")
+    vault._auth_github()  # noqa: SLF001
+else:
+    vault = Vault(
+        vault_addr=VAULT_ADDRESS, vault_role="dagster-server", aws_auth_mount="aws"
+    )
+    vault._auth_aws_iam()  # noqa: SLF001
+
+
+deployment_name = "openmetadata"
+deployment_assets = [
+    trino_metadata,
+]
+
+create_repository_using_definitions_args(
+    name=f"{deployment_name}_assets",
+    resources={
+        "vault": vault,
+        "s3": S3Resource(),
+    },
+    assets=deployment_assets,
+)

src/ol_orchestrate/partitions/openmetadata.py

@@ -0,0 +1,4 @@
+from dagster import DailyPartitionsDefinition
+
+# Define daily partitions
+TRINO_DAILY_PARTITIONS = DailyPartitionsDefinition(start_date="2023-01-01")

src/ol_orchestrate/resources/openmetadata/trino.py

@@ -0,0 +1,44 @@
+# treat this as a dictionary and reference variables in the appropriate locations
+trino_config = {
+    "source": {
+        "type": "trino",
+        "serviceName": "Starburst Galaxy",
+        "serviceConnection": {
+            "config": {
+                "type": "Trino",
+                "hostPort": "mitol-ol-data-lake-production.trino.galaxy.starburst.io:443",  # noqa: E501
+                "username": "[email protected]",
+                "authType": {
+                    "password": "<retrieve this from https://vault-production.odl.mit.edu/ui/vault/secrets/platform-secrets/kv/starburst-galaxy/details?version=4>"
+                },
+                "catalog": "ol_data_lake_production",
+            }
+        },
+        "sourceConfig": {
+            "config": {
+                "type": "DatabaseMetadata",
+                "markDeletedTables": True,
+                "markDeletedStoredProcedures": True,
+                "includeTables": True,
+                "includeViews": True,
+                "includeTags": True,
+                "includeOwners": True,
+                "includeStoredProcedures": True,
+                "includeDDL": True,
+                "threads": 10,
+            }
+        },
+    },
+    "sink": {"type": "metadata-rest", "config": {}},
+    "workflowConfig": {
+        "loggerLevel": "DEBUG",
+        "openMetadataServerConfig": {
+            "hostPort": "https://open-metadata-qa.ol.mit.edu/api",
+            "authProvider": "openmetadata",
+            "securityConfig": {
+                "jwtToken": "<retrieve this from https://open-metadata-ci.ol.mit.edu/bots/ingestion-bot>"
+            },
+            "storeServiceConnection": True,
+        },
+    },
+}

src/ol_orchestrate/workspace.yaml

@@ -6,5 +6,6 @@ load_from:
 - python_module: ol_orchestrate.definitions.edx.sync_program_credential_reports
 - python_module: ol_orchestrate.definitions.lakehouse.elt
 - python_module: ol_orchestrate.definitions.platform.notification
+- python_module: ol_orchestrate.definitions.platform.openmetadata
 - python_module: ol_orchestrate.repositories.edx_gcs_courses
 - python_module: ol_orchestrate.repositories.open_edx