V14 Publish dbt output

Use your current course session. Preparation and course runner · R script ↓

Optional deep dive. Enable only the options named in each block, after preparing its packages or services.

V14a Read a dbt result as a model

Goal: Bring SQL output into R.

Prerequisites: V13.

Reopen the R connection only after the CLI process finishes.

training_step(
  "V14a",
  {
    dbt_lake <- dataraft.lake::dr_connect_lake(dbt_config)
    dbt_model_id <- "model.training_policies.customer_revenue"
    dbt_dm <- dataraft.dbt::dr_dbt_model(
      dbt_lake,
      dbt_result,
      tables = c(premiums = dbt_model_id),
      primary_keys = list(premiums = "customer_id")
    )
    premium_totals <- dplyr::collect(dbt_dm$premiums)
    print(premium_totals)
    stopifnot(sum(premium_totals$revenue) == 740)
  },
  needs = c("V13"),
  enabled = training_options$dbt,
  optional = TRUE
)

Expected result: Customer premium totals of 250, 280 and 210. revenue is the template's technical output name.

Task: What is the output grain?

V14b Derive a dbt output contract

Goal: Review and version the output schema.

Prerequisites: V14a.

A manifest only contains its declared metadata. For this teaching product, confirm a contract derived from the actual result.

training_step(
  "V14b",
  {
    dbt_output_contract <- dr_contract_from(
      premium_totals,
      "customer.premiums",
      grain = "One customer",
      key = "customer_id",
      required = c("customer_id", "revenue")
    ) |>
      dr_contract_confirm()
    dbt_contract_spec <- dataraft.dbt::dr_dbt_contract(
      dbt_output_contract,
      name = "customer_revenue"
    )
    print(dbt_contract_spec)
  },
  needs = c("V14a"),
  enabled = training_options$dbt,
  optional = TRUE
)

Expected result: A contract matching the aggregated table.

Task: Are inferred types a business approval?

V14 Publish dbt output

Goal: Expose SQL results as a managed release.

Prerequisites: V14b.

Publication links the dbt artifact to the actual output table.

training_step(
  "V14",
  {
    dbt_release <- dataraft.dbt::dr_dbt_publish(
      dbt_lake,
      dbt_result,
      dbt_model_id,
      contract = dbt_output_contract,
      asset = "customer.premiums",
      code_version = "training-dbt-v1"
    )
    print(dr_collect(dbt_release))
    stopifnot(dbt_release$status == "published")
    dataraft.lake::dr_disconnect_lake(dbt_lake)
  },
  needs = c("V14b"),
  enabled = training_options$dbt,
  optional = TRUE
)

Expected result: A managed release of customer premium totals.

Task: Is dr_transform_dbt the same workflow?

V14c Inspect a manifest contract

Goal: Read declared dbt metadata.

Prerequisites: V13a.

Missing types in a manifest can prevent derivation. Show the concrete error instead of inventing a schema.

training_step(
  "V14c",
  {
    manifest_contract <- tryCatch(
      dataraft.dbt::dr_dbt_contract_from_manifest(
        dbt_result,
        "model.training_policies.customer_revenue"
      ),
      error = identity
    )
    print(manifest_contract)
  },
  needs = c("V13a"),
  enabled = training_options$dbt,
  optional = TRUE
)

Expected result: A contract or a specific diagnostic about missing or unsuitable declarations.

Task: How do you fix a missing type declaration?

V14d Use dbt as a transformation adapter

Goal: Bind your own prepared model.

Prerequisites: dbt=TRUE; TRAINING_DBT_TRANSFORM_PROJECT, TRAINING_DBT_TRANSFORM_MODEL and TRAINING_DBT_TRANSFORM_DB.

The model must read training_input in the specified DuckDB. This exercise does not silently rewrite the starter. Provide a separate test project with that interface and profile.

training_step(
  "V14d",
  {
    transform_path <- Sys.getenv("TRAINING_DBT_TRANSFORM_PROJECT")
    transform_model <- Sys.getenv("TRAINING_DBT_TRANSFORM_MODEL")
    transform_db <- Sys.getenv("TRAINING_DBT_TRANSFORM_DB")
    stopifnot(
      nzchar(transform_path),
      nzchar(transform_model),
      nzchar(transform_db)
    )
    transform_factory <- function() {
      DBI::dbConnect(duckdb::duckdb(), transform_db)
    }
    {
      transform_project <- dataraft.dbt::dr_dbt_project(
        transform_path,
        profiles_dir = transform_path,
        executable = Sys.getenv("TRAINING_DBT_EXECUTABLE", "dbt")
      )
      dbt_transform <- dataraft.dbt::dr_transform_dbt(
        transform_project,
        model = transform_model,
        connection = transform_factory,
        input = "training_input"
      )
      print(dr_collect(dr_run(
        dr_product("policies.dbt.transform", policies) |>
          dr_add_recipe(dr_recipe() |> dr_step_transform(dbt_transform)),
        write = FALSE
      )))
    }
  },
  packages = c("dataraft.dbt", "duckdb", "processx"),
  enabled = training_options$dbt &&
    nzchar(Sys.getenv("TRAINING_DBT_TRANSFORM_PROJECT")),
  optional = TRUE
)

Expected result: The configured model output. write=FALSE does not prevent dbt's own side effects.

Task: Why is this excluded from the default run?

V14e Deliver dbt artifacts to OpenMetadata

Goal: Send models, tests and SQL lineage through ingestion.

Prerequisites: V13a; catalogs=TRUE; prepared OpenMetadata database service; TRAINING_OPENMETADATA_URL, TRAINING_OPENMETADATA_SERVICE, OPENMETADATA_JWT_TOKEN; metadata CLI in PATH or TRAINING_METADATA_EXECUTABLE.

The service and its tables must already exist. Refresh database ingestion after adding RAW relations. delivered means CLI success, not proof that every expected edge exists.

training_step(
  "V14e",
  {
    om_dbt_url <- Sys.getenv("TRAINING_OPENMETADATA_URL")
    om_service <- Sys.getenv("TRAINING_OPENMETADATA_SERVICE")
    stopifnot(
      nzchar(om_dbt_url),
      nzchar(om_service),
      nzchar(Sys.getenv("OPENMETADATA_JWT_TOKEN"))
    )
    om_dbt <- dataraft.adapters::dr_catalog_openmetadata_dbt(
      om_dbt_url,
      service = om_service,
      executable = Sys.getenv("TRAINING_METADATA_EXECUTABLE", "metadata")
    )
    om_receipt <- dr_publish_metadata(om_dbt, dbt_result)
    print(om_receipt)
  },
  needs = c("V13a"),
  packages = c("dataraft.adapters", "processx"),
  enabled = training_options$catalogs && training_options$dbt,
  optional = TRUE
)

Expected result: A delivery receipt, or pending/blocked with diagnostics.

Task: Which prerequisite should you check if an edge is missing despite CLI success?