Integrating DBT with Superset

src/ol_orchestrate/definitions/lakehouse/elt.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 
 from dagster import (
+    AssetExecutionContext,
     AssetSelection,
     AutoMaterializePolicy,
     DefaultScheduleStatus,
@@ -13,8 +14,8 @@
 )
 from dagster_airbyte import airbyte_resource, load_assets_from_airbyte_instance
 from dagster_dbt import (
-    dbt_cli_resource,
-    load_assets_from_dbt_manifest,
+    DbtCliResource,
+    dbt_assets,
 )
 
 dagster_deployment = os.getenv("DAGSTER_ENVIRONMENT", "dev")
@@ -42,7 +43,7 @@
     "profiles_dir": str(dbt_repo_dir),
     "target": dagster_deployment,
 }
-configured_dbt_cli = dbt_cli_resource.configured(dbt_config)
+configured_dbt_cli = DbtCliResource(dbt_config)
 
 
 def filter_active_connections(connection) -> bool:
@@ -79,11 +80,16 @@ def filter_active_connections(connection) -> bool:
     default_status=DefaultScheduleStatus.RUNNING,
 )
 
-dbt_assets = load_assets_from_dbt_manifest(
+
+@dbt_assets(
     manifest=json.loads(
         dbt_repo_dir.joinpath("target", "manifest.json").read_text(),
     ),
 )
+def dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
+    """A special asset that will load all dbt assets."""
+    yield from dbt.cli(["build"], context=context).stream()
+
 
 elt = Definitions(
     assets=load_assets_from_current_module(

src/ol_orchestrate/jobs/open_edx.py

@@ -13,7 +13,9 @@
     export_edx_forum_database,
     fetch_edx_course_structure_from_api,
     list_courses,
+    push_dbt_metadata,
     student_submissions,
+    sync_superset_data,
     upload_extracted_data,
     user_roles,
     write_course_list_csv,
@@ -53,6 +55,9 @@ def edx_course_pipeline():
             notify_healthchecks_io_on_failure,
         }
     )(course_list, extracts_upload)
+    # TODO pass a list of dbt tables and superset tables
+    sync_superset_data(dbt_tables_marts, superset_tables)
+    push_dbt_metadata()
 
 
 @graph(

src/ol_orchestrate/ops/open_edx.py

@@ -79,6 +79,15 @@ class UploadExtractedDataConfig(Config):
         description="S3 bucket to use for uploading results of pipeline execution.",
     )
 
+class SyncSupersetDataConfig(Config):
+    tables_to_sync: list[str] = Field(
+        default=["marts",],
+        description="Set of tables to sync in Superset.",
+    )
+
+class PushDbtMetadataConfig(Config):
+    pass
+
 
 @op(
     name="list_edx_courses",
@@ -737,3 +746,42 @@ def upload_extracted_data(
         f"{results_bucket}/{context.resources.results_dir.path.name}",
         "edx_daily_extracts_directory",
     )
+
+@op(
+    name="sync_superset_dbt_datasets",
+    description="Adds new reporting tables to corresponding schemas,"
+                "drops superset tables that are no longer present in dbt,"
+                "automates column refresh of physical datasets to keep them updated.",
+    required_resource_keys={"s3",},
+    ins={},
+    out={},
+)
+def sync_superset_data(
+    context: OpExecutionContext,
+    config: SyncSupersetDataConfig,
+    dbt_tables_marts: list[str],
+    superset_tables: list[str],
+):
+# Parsing as sets
+dbt_tables_marts=set(dbt_tables_marts)
+superset_tables=set(superset_tables)
+
+# Add missing tables
+add_to_superset=list(dbt_tables_marts.difference(superset_tables))
+len(add_to_superset)
+
+# Remove extra tables
+remove_from_superset=list(superset_tables.difference(dbt_tables_marts))
+len(remove_from_superset)
+
+@op(
+    name="push_dbt_metadata_to_superset",
+    description="Pushes all dbt documentation including column and table descriptionns to superset",  # noqa: E501
+    required_resource_keys={},
+    ins={},
+    out={},
+)
+def push_dbt_metadata(
+    context: OpExecutionContext,
+    config: PushDbtMetadataConfig,
+):