Dashboard

The final step in our pipeline is configuring our dashboard. Including the presentation layer in our project helps ensure the full lineage of our project and ensures that data will be updated as soon as it is transformed. This is much easier than trying to coordinate schedules across services.

For this example we will use Power BI but Dagster has native support for BI tools such as Tableau, Looker and Sigma. And like the native dbt resource, we will not need to add much code to include our PowerBI assets.

First we will initialize the PowerBIWorkspace resource which allows Dagster to communicate with Power BI.

power_bi_workspace = PowerBIWorkspace(
    credentials=PowerBIServicePrincipal(
        client_id=dg.EnvVar("AZURE_POWERBI_CLIENT_ID"),
        client_secret=dg.EnvVar("AZURE_POWERBI_CLIENT_SECRET"),
        tenant_id=dg.EnvVar("AZURE_POWERBI_TENANT_ID"),
    ),
    workspace_id=dg.EnvVar("AZURE_POWERBI_WORKSPACE_ID"),
)

Then, like dbt, we will define a translator. This time since the Power BI assets live downstream of our dbt models, we will map the Power BI assets to those model assets.

class CustomDagsterPowerBITranslator(DagsterPowerBITranslator):
    def get_report_spec(self, data: PowerBITranslatorData) -> dg.AssetSpec:
        return (
            super()
            .get_report_spec(data)
            .replace_attributes(
                group_name="reporting",
            )
        )

    def get_semantic_model_spec(self, data: PowerBITranslatorData) -> dg.AssetSpec:
        upsteam_table_deps = [
            dg.AssetKey(table.get("name")) for table in data.properties.get("tables", [])
        ]
        return (
            super()
            .get_semantic_model_spec(data)
            .replace_attributes(
                group_name="reporting",
                deps=upsteam_table_deps,
            )
        )

Finally we define the definition for our dashboard assets and Power BI resource.

power_bi_specs = load_powerbi_asset_specs(
    power_bi_workspace,
    dagster_powerbi_translator=CustomDagsterPowerBITranslator(),
)

defs = dg.Definitions(assets=[*power_bi_specs], resources={"power_bi": power_bi_workspace})

Definition merge

With the dashboard definition set, we have all three layers of the end-to-end project ready to go. We can now define a single definition which will be the definition used in our code location.

import project_atproto_dashboard.dashboard.definitions as dashboard_definitions
import project_atproto_dashboard.ingestion.definitions as ingestion_definitions
import project_atproto_dashboard.modeling.definitions as modeling_definitions

defs = dg.Definitions.merge(
    ingestion_definitions.defs, modeling_definitions.defs, dashboard_definitions.defs
)

You can see that organizing your project into domain specific definitions leads to a clean definition. We do this with our own internal Dagster project that combines over a dozen domain specific definitions for the various tools and services we use.

Definition merge​

Definition merge