diff --git a/elt-pipelines/README.md b/elt-pipelines/README.md index 658469c4..ddccbf5e 100644 --- a/elt-pipelines/README.md +++ b/elt-pipelines/README.md @@ -104,11 +104,14 @@ elt-pipelines/ | | |-- / | | | |-- / | | | | |-- .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 `_` namespace +- The `transform` directory contains a `dbt` project which defines ways of transforming the raw data in the landing + warehouse into cleaned 'models' in ``. For more details see + [one of the project readmes](facility_ops/transform/README.md) diff --git a/elt-pipelines/facility_ops/transform/.gitignore b/elt-pipelines/facility_ops/transform/.gitignore new file mode 100644 index 00000000..64b8f622 --- /dev/null +++ b/elt-pipelines/facility_ops/transform/.gitignore @@ -0,0 +1,8 @@ +logs/ +target/ +dbt_packages/ +spark-warehouse/ +metastore_db/ + +derby.log +.user.yml diff --git a/elt-pipelines/facility_ops/transform/.sqlfluff b/elt-pipelines/facility_ops/transform/.sqlfluff new file mode 100644 index 00000000..72890088 --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/.sqlfluffignore b/elt-pipelines/facility_ops/transform/.sqlfluffignore new file mode 100644 index 00000000..7b96c194 --- /dev/null +++ b/elt-pipelines/facility_ops/transform/.sqlfluffignore @@ -0,0 +1,4 @@ +reports +target +dbt_packages +macros diff --git a/elt-pipelines/facility_ops/transform/README.md b/elt-pipelines/facility_ops/transform/README.md new file mode 100644 index 00000000..d25debc8 --- /dev/null +++ b/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 --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 diff --git a/elt-pipelines/facility_ops/transform/analyses/.gitkeep b/elt-pipelines/facility_ops/transform/analyses/.gitkeep new file mode 100644 index 00000000..e69de29b diff --git a/elt-pipelines/facility_ops/transform/dbt_project.yml b/elt-pipelines/facility_ops/transform/dbt_project.yml new file mode 100644 index 00000000..b1c293cb --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/macros/create_equipment_category_key.sql b/elt-pipelines/facility_ops/transform/macros/create_equipment_category_key.sql new file mode 100644 index 00000000..6b5aaba2 --- /dev/null +++ b/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 -%} diff --git a/elt-pipelines/facility_ops/transform/macros/normalize_whitespace.sql b/elt-pipelines/facility_ops/transform/macros/normalize_whitespace.sql new file mode 100644 index 00000000..8d0bab94 --- /dev/null +++ b/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 -%} diff --git a/elt-pipelines/facility_ops/transform/macros/parse_utc_timestamp.sql b/elt-pipelines/facility_ops/transform/macros/parse_utc_timestamp.sql new file mode 100644 index 00000000..84bdfaef --- /dev/null +++ b/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 -%} diff --git a/elt-pipelines/facility_ops/transform/models/marts/accelerator/power_consumption.sql b/elt-pipelines/facility_ops/transform/models/marts/accelerator/power_consumption.sql new file mode 100644 index 00000000..b849fda4 --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/models/marts/accelerator/power_consumption.yml b/elt-pipelines/facility_ops/transform/models/marts/accelerator/power_consumption.yml new file mode 100644 index 00000000..fe492fd1 --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/models/staging/estates/_estates__sources.yml b/elt-pipelines/facility_ops/transform/models/staging/estates/_estates__sources.yml new file mode 100644 index 00000000..cafd7824 --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/models/staging/estates/stg_electricity_sharepoint_rdm_data.sql b/elt-pipelines/facility_ops/transform/models/staging/estates/stg_electricity_sharepoint_rdm_data.sql new file mode 100644 index 00000000..0bab0a59 --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/models/staging/estates/stg_electricity_sharepoint_rdm_data.yml b/elt-pipelines/facility_ops/transform/models/staging/estates/stg_electricity_sharepoint_rdm_data.yml new file mode 100644 index 00000000..6a193366 --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/package-lock.yml b/elt-pipelines/facility_ops/transform/package-lock.yml new file mode 100644 index 00000000..017c94e5 --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/packages.yml b/elt-pipelines/facility_ops/transform/packages.yml new file mode 100644 index 00000000..f4bc168a --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/profiles.yml b/elt-pipelines/facility_ops/transform/profiles.yml new file mode 100644 index 00000000..115f8367 --- /dev/null +++ b/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 }}" diff --git a/warehouses/facility_ops/transform/requirements/developer.in b/elt-pipelines/facility_ops/transform/requirements/developer.in similarity index 100% rename from warehouses/facility_ops/transform/requirements/developer.in rename to elt-pipelines/facility_ops/transform/requirements/developer.in diff --git a/warehouses/facility_ops/transform/requirements/developer.txt b/elt-pipelines/facility_ops/transform/requirements/developer.txt similarity index 100% rename from warehouses/facility_ops/transform/requirements/developer.txt rename to elt-pipelines/facility_ops/transform/requirements/developer.txt diff --git a/warehouses/facility_ops/transform/requirements/requirements.in b/elt-pipelines/facility_ops/transform/requirements/requirements.in similarity index 100% rename from warehouses/facility_ops/transform/requirements/requirements.in rename to elt-pipelines/facility_ops/transform/requirements/requirements.in diff --git a/warehouses/facility_ops/transform/requirements/requirements.txt b/elt-pipelines/facility_ops/transform/requirements/requirements.txt similarity index 100% rename from warehouses/facility_ops/transform/requirements/requirements.txt rename to elt-pipelines/facility_ops/transform/requirements/requirements.txt diff --git a/elt-pipelines/facility_ops/transform/seeds/.gitkeep b/elt-pipelines/facility_ops/transform/seeds/.gitkeep new file mode 100644 index 00000000..e69de29b diff --git a/elt-pipelines/facility_ops/transform/snapshots/.gitkeep b/elt-pipelines/facility_ops/transform/snapshots/.gitkeep new file mode 100644 index 00000000..e69de29b diff --git a/elt-pipelines/facility_ops/transform/tests/.gitkeep b/elt-pipelines/facility_ops/transform/tests/.gitkeep new file mode 100644 index 00000000..e69de29b diff --git a/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__additional_columns.sql b/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__additional_columns.sql new file mode 100644 index 00000000..ee716ee4 --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__chapter_entry.sql b/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__chapter_entry.sql new file mode 100644 index 00000000..73b26b90 --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__entries.sql b/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__entries.sql new file mode 100644 index 00000000..c29f1759 --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__logbook_chapter.sql b/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__logbook_chapter.sql new file mode 100644 index 00000000..d7c3f375 --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__logbooks.sql b/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__logbooks.sql new file mode 100644 index 00000000..30e4e40b --- /dev/null +++ b/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 diff --git a/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__more_entry_columns.sql b/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__more_entry_columns.sql new file mode 100644 index 00000000..662fa7eb --- /dev/null +++ b/elt-pipelines/facility_ops/transform/tests/fixtures/base_opralogweb__more_entry_columns.sql @@ -0,0 +1,78 @@ +select + + cast(100 as integer) as entry_id, + cast('Equipment 100' as varchar) as string_data, + cast(null as double) as number_data, + cast(1 as integer) as additional_column_id + +union all + +select + + cast(100 as integer) as entry_id, + cast('Group 100' as varchar) as string_data, + cast(null as double) as number_data, + cast(2 as integer) as additional_column_id + +union all + +select + + cast(100 as integer) as entry_id, + cast(null as varchar) as string_data, + cast(4.2 as double) as number_data, + cast(3 as integer) as additional_column_id + +union all + +select + + cast(101 as integer) as entry_id, + cast('Equipment 101' as varchar) as string_data, + cast(null as double) as number_data, + cast(1 as integer) as additional_column_id + +union all + +select + + cast(101 as integer) as entry_id, + cast('Group 101' as varchar) as string_data, + cast(null as double) as number_data, + cast(2 as integer) as additional_column_id + +union all + +select + + cast(101 as integer) as entry_id, + cast(null as varchar) as string_data, + cast(5.3 as double) as number_data, + cast(3 as integer) as additional_column_id + +union all + +select + + cast(102 as integer) as entry_id, + cast('Equipment 102' as varchar) as string_data, + cast(null as double) as number_data, + cast(1 as integer) as additional_column_id + +union all + +select + + cast(102 as integer) as entry_id, + cast('Group 102' as varchar) as string_data, + cast(null as double) as number_data, + cast(2 as integer) as additional_column_id + +union all + +select + + cast(102 as integer) as entry_id, + cast(null as varchar) as string_data, + cast(4.3 as double) as number_data, + cast(3 as integer) as additional_column_id diff --git a/elt-pipelines/facility_ops/transform/tests/fixtures/base_statusdisplay__cycles.sql b/elt-pipelines/facility_ops/transform/tests/fixtures/base_statusdisplay__cycles.sql new file mode 100644 index 00000000..9eac95b0 --- /dev/null +++ b/elt-pipelines/facility_ops/transform/tests/fixtures/base_statusdisplay__cycles.sql @@ -0,0 +1,4 @@ +select + + cast('h344o234ii5o6o4' as varchar) as dlt_id, + cast('2024/2' as varchar) as name diff --git a/elt-pipelines/facility_ops/transform/tests/fixtures/base_statusdisplay__cycles__phases.sql b/elt-pipelines/facility_ops/transform/tests/fixtures/base_statusdisplay__cycles__phases.sql new file mode 100644 index 00000000..5194a88e --- /dev/null +++ b/elt-pipelines/facility_ops/transform/tests/fixtures/base_statusdisplay__cycles__phases.sql @@ -0,0 +1,7 @@ +select + + cast('user-time' as varchar) as phase, + cast(1 as integer) as target, + timestamp '2024-07-09 07:30:00 UTC' as started_at, + timestamp '2024-07-24 07:30:00 UTC' as ended_at, + cast('h344o234ii5o6o4' as varchar) as dlt_cycles_id