feat(elt-pipelines): dbt project with electricity_sharepoint transform

elt-pipelines/README.md

@@ -104,11 +104,14 @@ elt-pipelines/
 |    |    |-- <domain>/
 |    |    |    |-- <job name>/
 |    |    |    |    |-- <job name>.py
+|    |-- transform/
 ```
 
 - This directory structure is required for using `elt-common`
-- Each 'target warehouse' is the name of an Iceberg warehouse. The data ingested by the pipelines inside that directory
-  end up in that warehouse.
-- Data from ingest pipelines is considered 'raw' data, and is loaded into a warehouse suffixed with `_landing`.
-- Under construction: Each warehouse will also have a `transform` subdirectory containing pipelines for converting the
-  raw data into its final state in the target warehouse.
+- Each 'target warehouse' is the name of an Iceberg warehouse. The data handled by the pipelines inside that directory
+  end up in that warehouse
+- Data from ingest pipelines is considered 'raw' data, and is loaded into a warehouse suffixed with `_landing` under
+  the `<domain>_<job name>` namespace
+- The `transform` directory contains a `dbt` project which defines ways of transforming the raw data in the landing
+  warehouse into cleaned 'models' in `<target warehouse>`. For more details see
+  [one of the project readmes](facility_ops/transform/README.md)

elt-pipelines/facility_ops/transform/.gitignore

@@ -0,0 +1,8 @@
+logs/
+target/
+dbt_packages/
+spark-warehouse/
+metastore_db/
+
+derby.log
+.user.yml

elt-pipelines/facility_ops/transform/.sqlfluff

@@ -0,0 +1,37 @@
+[sqlfluff]
+dialect = trino
+templater = dbt
+runaway_limit = 10
+max_line_length = 80
+indent_unit = space
+
+[sqlfluff:indentation]
+tab_space_size = 4
+
+[sqlfluff:layout:type:comma]
+spacing_before = touch
+line_position = trailing
+
+[sqlfluff:rules:capitalisation.keywords]
+capitalisation_policy = lower
+
+[sqlfluff:rules:aliasing.table]
+aliasing = explicit
+
+[sqlfluff:rules:aliasing.column]
+aliasing = explicit
+
+[sqlfluff:rules:aliasing.expression]
+allow_scalar = False
+
+[sqlfluff:rules:capitalisation.identifiers]
+extended_capitalisation_policy = lower
+
+[sqlfluff:rules:capitalisation.functions]
+capitalisation_policy = lower
+
+[sqlfluff:rules:capitalisation.literals]
+capitalisation_policy = lower
+
+[sqlfluff:rules:ambiguous.column_references]  # Number in group by
+group_by_and_order_by_style = implicit

elt-pipelines/facility_ops/transform/.sqlfluffignore

@@ -0,0 +1,4 @@
+reports
+target
+dbt_packages
+macros

elt-pipelines/facility_ops/transform/README.md

@@ -0,0 +1,30 @@
+# Facility Operations catalog models
+
+This is a [dbt](https://docs.getdbt.com/) project which defines transforms for turning the
+raw data in the `facility_ops_landing` warehouse into cleaned models in `facility_ops`.
+
+**Under construction. Currently only the electricity_sharepoint transforms work.**
+
+## Running with elt
+
+The `elt run` command defined by [`elt-common`](../../../elt-common) can be used to run
+transforms from this project.
+
+Ingest jobs ingest data into a namespace defined by the
+[ingest directory structure](../../README.md#directory-structure). That namespace can be used
+to run the transform(s) for the pipeline with `elt run facility_ops <namespace> --step transform`.
+
+## Running with dbt
+
+It's also possible to run the transforms (or interact with them in other ways) using `dbt` directly.
+
+- Use a python environment with [elt-pipelines](../../README.md#setting-up-a-python-virtual-environment)
+  or [elt-common](../../../elt-common/README.md#setting-up-a-python-virtual-environment) installed
+   - These provide the required `dbt` dependencies, and the `dbt` cli tool
+- Make the dbt project directory (`elt-pipelines/facility_ops/transform`) the working directory
+- Run `dbt deps` to install the project dependencies
+- For running against a local catalog, ensure the docker
+  services [are running](../../../infra/local/README.md#local-set-up)
+- Run `dbt` commands whilst in the dbt project directory
+- To run against a remote catalog, ensure the [required environment variables](./profiles.yml) are set up to point to
+  the Trino instance, then use the `--profile remote` option when running `dbt` commands

elt-pipelines/facility_ops/transform/dbt_project.yml

@@ -0,0 +1,44 @@
+name: facility_ops
+version: "1.0.0"
+
+# default project points at stack in local/infra
+profile: "local"
+
+flags:
+  require_certificate_validation: True
+
+model-paths: [ "models" ]
+analysis-paths: [ "analyses" ]
+test-paths: [ "tests" ]
+seed-paths: [ "seeds" ]
+macro-paths: [ "macros" ]
+snapshot-paths: [ "snapshots" ]
+
+clean-targets:
+  # directories to be removed by `dbt clean`
+  - "target"
+  - "dbt_packages"
+
+dispatch:
+  # Use macros from trino_utils in preference. https://hub.getdbt.com/starburstdata/trino_utils/latest/
+  - macro_namespace: dbt_utils
+    search_order: [ "trino_utils", "dbt_utils" ]
+  - macro_namespace: dbt_date
+    search_order: [ "trino_utils", "dbt_date" ]
+  - macro_namespace: metrics
+    search_order: [ "trino_utils", "metrics" ]
+
+models:
+  +file_format: parquet
+  +materialized: view
+  facility_ops:
+    # The full schema name of the final tables is "{target_schema}_"{custom_schema}"
+    #   - 'target_schema' is defined in profiles.yml
+    #   - 'custom_schema' is defined for each collection of models below
+    # See https://docs.getdbt.com/docs/build/custom-schemas
+    staging:
+      +schema: staging
+    marts:
+      +materialized: table
+      accelerator:
+        +schema: accelerator

elt-pipelines/facility_ops/transform/macros/create_equipment_category_key.sql

@@ -0,0 +1,4 @@
+-- Creates a key for the equipment -> category mapping table
+{%- macro create_equipment_category_key(text_col) -%}
+  lower( {{ normalize_whitespace(text_col) }} )
+{%- endmacro -%}

elt-pipelines/facility_ops/transform/macros/normalize_whitespace.sql

@@ -0,0 +1,8 @@
+-- Removes leading and trailing whitespace and replaces any multiple-spaces with a single space
+{%- macro normalize_whitespace(text_col) -%}
+  {{ return(adapter.dispatch('normalize_whitespace')(text_col)) }}
+{% endmacro %}
+
+{%- macro default__normalize_whitespace(text_col) -%}
+  regexp_replace(trim({{ text_col }}), '\s+', ' ')
+{%- endmacro -%}

elt-pipelines/facility_ops/transform/macros/parse_utc_timestamp.sql

@@ -0,0 +1,16 @@
+-- Parse separate date and time strings into a single UTC timestamp
+{%- macro parse_utc_timestamp(date_col, date_format, time_col, time_format='HH:mm:ss ZZZ', src_timezone='UTC') -%}
+  {{ return(adapter.dispatch('parse_utc_timestamp')(date_col, date_format, time_col, time_format, src_timezone)) }}
+{% endmacro %}
+
+
+{%- macro trino__parse_utc_timestamp(date_col, date_format, time_col, time_format, src_timezone) -%}
+  with_timezone(
+    cast(
+      parse_datetime({{ adapter.quote(date_col) }} || ' ' || {{ adapter.quote(time_col) }} || ' {{ src_timezone }}',
+                     '{{ date_format ~ ' ' ~ time_format }}')
+                     as timestamp(3)
+        ),
+    '{{ src_timezone }}'
+  ) at time zone 'UTC'
+{%- endmacro -%}

elt-pipelines/facility_ops/transform/models/marts/accelerator/power_consumption.sql

@@ -0,0 +1,23 @@
+{{
+  config(
+    properties={
+      "partitioning": "ARRAY['year(power_measured_at)']",
+    },
+    on_table_exists = 'drop'
+)
+}}
+
+with
+
+staged as (
+
+  select
+
+    power_measured_at,
+    total_isis_power_mw
+
+  from {{ ref('stg_electricity_sharepoint_rdm_data') }}
+
+)
+
+select * from staged

elt-pipelines/facility_ops/transform/models/marts/accelerator/power_consumption.yml

@@ -0,0 +1,11 @@
+models:
+  - name: power_consumption
+    description: >
+      The total power consumption for ISIS over time.
+    columns:
+      - name: power_measured_at
+        data_tests:
+          - not_null
+      - name: total_isis_power_mw
+        data_tests:
+          - not_null

elt-pipelines/facility_ops/transform/models/staging/estates/_estates__sources.yml

@@ -0,0 +1,7 @@
+version: 2
+
+sources:
+  - name: estates_electricity_sharepoint
+    database: facility_ops_landing
+    tables:
+      - name: rdm_data

elt-pipelines/facility_ops/transform/models/staging/estates/stg_electricity_sharepoint_rdm_data.sql

@@ -0,0 +1,18 @@
+with source as (
+
+    select * from {{ source('estates_electricity_sharepoint', 'rdm_data') }}
+
+),
+
+renamed as (
+
+    select
+
+      date_time as power_measured_at,
+      isis_elec_total_power_mw as total_isis_power_mw
+
+    from source
+
+)
+
+select * from renamed

elt-pipelines/facility_ops/transform/models/staging/estates/stg_electricity_sharepoint_rdm_data.yml

@@ -0,0 +1,11 @@
+models:
+  - name: stg_electricity_sharepoint_rdm_data
+    description: >
+      ISIS electricity consumption.
+    columns:
+      - name: power_measured_at
+        data_tests:
+          - not_null
+      - name: total_isis_power_mw
+        data_tests:
+          - not_null

elt-pipelines/facility_ops/transform/package-lock.yml

@@ -0,0 +1,11 @@
+packages:
+  - name: codegen
+    package: dbt-labs/codegen
+    version: 0.14.0
+  - name: dbt_utils
+    package: dbt-labs/dbt_utils
+    version: 1.3.3
+  - name: trino_utils
+    package: starburstdata/trino_utils
+    version: 0.6.0
+sha1_hash: dc8ade786ecde1c757d4fd39ba36eb7ddc99a6af

elt-pipelines/facility_ops/transform/packages.yml

@@ -0,0 +1,7 @@
+packages:
+  - package: dbt-labs/codegen
+    version: 0.14.0
+  - package: dbt-labs/dbt_utils
+    version: 1.3.3
+  - package: starburstdata/trino_utils
+    version: 0.6.0

elt-pipelines/facility_ops/transform/profiles.yml

@@ -0,0 +1,31 @@
+local:
+  target: trino
+  outputs:
+    trino:
+      type: trino
+      method: ldap
+      user: machine-infra
+      password: s3cr3t
+      host: localhost
+      port: 58443
+      http_scheme: https
+      database: facility_ops
+      schema: analytics
+      threads: 8
+      cert: false
+      suppress_cert_warning: true
+
+remote:
+  target: trino
+  outputs:
+    trino:
+      type: trino
+      method: ldap
+      http_scheme: https
+      user: "{{ env_var('DBT_TRINO_USER') }}"
+      password: "{{ env_var('DBT_TRINO_PASSWORD') }}"
+      host: "{{ env_var('DBT_TRINO_HOST') }}"
+      port: "{{ env_var('DBT_TRINO_PORT') | int }}"
+      database: "{{ env_var('DBT_TRINO_CATALOG') }}"
+      schema: "{{ env_var('DBT_TRINO_CATALOG_SCHEMA_PREFIX') }}analytics"
+      threads: "{{ env_var('DBT_TRINO_THREADS', '8') | int }}"

elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__additional_columns.sql

@@ -0,0 +1,21 @@
+select
+  cast(1 as integer) as additional_column_id,
+  cast('Equipment' as varchar) as column_title
+
+union all
+
+select
+  cast(2 as integer) as additional_column_id,
+  cast('Group' as varchar) as column_title
+
+union all
+
+select
+  cast(3 as integer) as additional_column_id,
+  cast('Lost Time' as varchar) as column_title
+
+union all
+
+select
+  cast(4 as integer) as additional_column_id,
+  cast('Group Leader comments' as varchar) as column_title

elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__chapter_entry.sql

@@ -0,0 +1,24 @@
+select
+
+  cast(100 as integer) as entry_id,
+  cast(24 as integer) as principal_logbook,
+  cast(1 as integer) as logbook_chapter_no,
+  cast(24 as integer) as logbook_id
+
+union all
+
+select
+
+  cast(101 as integer) as entry_id,
+  cast(24 as integer) as principal_logbook,
+  cast(2 as integer) as logbook_chapter_no,
+  cast(24 as integer) as logbook_id
+
+union all
+
+select
+
+  cast(102 as integer) as entry_id,
+  cast(24 as integer) as principal_logbook,
+  cast(2 as integer) as logbook_chapter_no,
+  cast(24 as integer) as logbook_id

elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__entries.sql

@@ -0,0 +1,27 @@
+select
+
+  cast(100 as integer) as entry_id,
+  with_timezone(timestamp '2017-04-25 23:59:59', 'UTC') as fault_occurred_at,
+  date '2017-04-24' as fault_date,
+  cast('Comment 100' as varchar) as fault_description,
+  false as logically_deleted
+
+union all
+
+select
+
+  cast(101 as integer) as entry_id,
+  with_timezone(timestamp '2024-01-01 00:00:01', 'UTC') as fault_occurred_at,
+  date '2024-01-01' as fault_date,
+  cast('Comment 101' as varchar) as fault_description,
+  false as logically_deleted
+
+union all
+
+select
+
+  cast(102 as integer) as entry_id,
+  with_timezone(timestamp '2024-01-01 00:01:03', 'UTC') as fault_occurred_at,
+  date '2024-01-01' as fault_date,
+  cast('Deleted 102' as varchar) as fault_description,
+  true as logically_deleted

elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__logbook_chapter.sql

@@ -0,0 +1,9 @@
+select
+
+  cast(1 as integer) as logbook_chapter_no
+
+union all
+
+select
+
+  cast(2 as integer) as logbook_chapter_no

elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__logbooks.sql

@@ -0,0 +1,4 @@
+select
+
+  cast(24 as integer) as logbook_id,
+  cast('MCR Running Log' as varchar) as logbook_name