From 1f39ed4336ff5aa740b1548d185300c6ee37d35c Mon Sep 17 00:00:00 2001 From: Martyn Gigg Date: Tue, 9 Dec 2025 16:28:51 +0000 Subject: [PATCH 1/3] Electricity ingest: Skip files that cannot be processed. Occasional gaps in the data is better than no data at all. --- .../dlt_destinations/pyiceberg/pyiceberg.py | 29 +++++++++--- .../pyiceberg/test_pyiceberg_pipeline.py | 46 +++++++++++++++++-- .../extract_and_load.py | 11 +++-- 3 files changed, 70 insertions(+), 16 deletions(-) diff --git a/elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py b/elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py index 055b8ed2..581a94ae 100644 --- a/elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py +++ b/elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py @@ -411,15 +411,30 @@ def write_to_table( # skip is internally used for dlt tables table.append(table_data) elif write_disposition == "merge": - strategy = self.prepare_load_table(table_name)["x-merge-strategy"] # type:ignore + dlt_schema = self.prepare_load_table(table_name) + join_columns = None + if not table.schema().identifier_field_ids: + join_columns = [ + name + for name, column in dlt_schema["columns"].items() + if column.get("merge_key", False) + ] + # provide merge key columns if no identifier-field-ids exist to mark the primary key columns + strategy = dlt_schema["x-merge-strategy"] # type:ignore if strategy == "upsert": # requires the indentifier fields to have been defined - table.upsert( - df=table_data, - when_matched_update_all=True, - when_not_matched_insert_all=True, - case_sensitive=True, - ) + try: + table.upsert( + df=table_data, + join_cols=join_columns, + when_matched_update_all=True, + when_not_matched_insert_all=True, + case_sensitive=True, + ) + except ValueError as exc: + logger.info(f"----- Existing data ------\n{table.scan().to_arrow()}") + logger.info(f"----- New data ------\n{table_data}") + raise exc else: raise DestinationTerminalException( f'Merge strategy "{strategy}" is not supported for Iceberg tables. ' diff --git a/elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py b/elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py index dc521051..db714506 100644 --- a/elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py +++ b/elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py @@ -33,9 +33,17 @@ def pipeline_name(frame: FrameType | None): return frame.f_code.co_name if frame is not None else "pipeline_name_frame_none" -def resource_factory(data: List[Dict[str, Any]] | None = None, primary_key: str | None = "id"): +def resource_factory( + data: List[Dict[str, Any]] | None = None, primary_key: str | None = "id", merge_key=None +): + kwargs = {} if primary_key is not None: - decorator = dlt.resource(primary_key=primary_key) + kwargs["primary_key"] = primary_key + if merge_key is not None: + kwargs["merge_key"] = merge_key + + if kwargs: + decorator = dlt.resource(**kwargs) else: decorator = dlt.resource @@ -168,10 +176,18 @@ def test_explicit_replace( ) +@pytest.mark.parametrize( + "identifier_keys", + [ + {"primary_key": "id", "merge_key": None}, + {"primary_key": None, "merge_key": "id"}, + ], +) def test_explicit_merge_updates_expected_values( warehouse: Warehouse, pipelines_dir, destination_config: PyIcebergDestinationTestConfiguration, + identifier_keys: dict, ) -> None: num_records_first_run = 1000 data = [ @@ -187,7 +203,7 @@ def test_explicit_merge_updates_expected_values( pipelines_dir=pipelines_dir, ) # first run - pipeline.run(resource_factory(data, primary_key="id")) + pipeline.run(resource_factory(data, **identifier_keys)) assert_table_has_data( pipeline, f"{pipeline.dataset_name}.data_items", @@ -204,7 +220,7 @@ def test_explicit_merge_updates_expected_values( } for i in range(num_records_upserted) ] - pipeline.run(resource_factory(data_updated, primary_key="id"), write_disposition="merge") + pipeline.run(resource_factory(data_updated, **identifier_keys), write_disposition="merge") expected_data = [ {"id": i + 1, "category": "A" if i + 1 < id_updated_start else "B"} for i in range(num_records_first_run + num_records_upserted - id_updated_start + 1) @@ -217,6 +233,28 @@ def test_explicit_merge_updates_expected_values( ) +def test_merge_without_primary_or_merge_key_raises_error( + warehouse: Warehouse, + pipelines_dir, + destination_config: PyIcebergDestinationTestConfiguration, +) -> None: + num_records_first_run = 1000 + data = [ + { + "id": i + 1, + "category": "A", + } + for i in range(num_records_first_run) + ] + pipeline = destination_config.setup_pipeline( + warehouse, + pipeline_name(inspect.currentframe()), + pipelines_dir=pipelines_dir, + ) + with pytest.raises(Exception): + pipeline.run(resource_factory(data, primary_key=None), write_disposition="merge") + + @pytest.mark.parametrize("merge_strategy", ["delete-insert", "scd2"]) def test_explicit_merge_not_supported_for_strategies_other_than_upsert( warehouse: Warehouse, diff --git a/warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py b/warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py index 56e4b142..a6cbe060 100644 --- a/warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py +++ b/warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py @@ -117,10 +117,11 @@ def read_as_dataframe(file_obj: M365DriveItem) -> pd.DataFrame | None: df = read_power_consumption_excel(io.BytesIO(file_bytes), skip_rows) case _: raise RuntimeError(f"Unsupported file extension in '{file_name}'") - except pd.errors.EmptyDataError as exc: - raise RuntimeError( - f"'{file_name} ({len(file_bytes)} bytes)': {str(exc)}" - ) from exc + except Exception as exc: + LOGGER.error( + f"'Error loading {file_name} ({len(file_bytes)} bytes)'. File skipped.\nDetails: {str(exc)}" + ) + return None df["file_name"] = file_obj["file_name"] return df @@ -140,7 +141,7 @@ def read_as_dataframe(file_obj: M365DriveItem) -> pd.DataFrame | None: yield df_batch -@dlt.resource(merge_key="file_name") +@dlt.resource(merge_key="DateTime", columns={"DateTime": {"nullable": False}}) def rdm_data( datetime_cur=dlt.sources.incremental( "DateTime", From 8dd95a606a4335857e419e68ecadcc03654948b2 Mon Sep 17 00:00:00 2001 From: Martyn Gigg Date: Tue, 9 Dec 2025 17:00:09 +0000 Subject: [PATCH 2/3] Move Trino port in local compose config It seems to now conflict with something on GitHub Actions. --- elt-common/tests/e2e_tests/conftest.py | 2 +- infra/local/docker-compose.yml | 4 ++-- warehouses/accelerator/transform/accelerator/profiles.yml | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/elt-common/tests/e2e_tests/conftest.py b/elt-common/tests/e2e_tests/conftest.py index a1599633..aac469db 100644 --- a/elt-common/tests/e2e_tests/conftest.py +++ b/elt-common/tests/e2e_tests/conftest.py @@ -92,7 +92,7 @@ class Settings(BaseSettings): # trino trino_http_scheme: str = "http" trino_host: str = "localhost" - trino_port: str = "58088" + trino_port: str = "59088" trino_user: str = "trino" trino_password: str = "" diff --git a/infra/local/docker-compose.yml b/infra/local/docker-compose.yml index 80740a9a..cd562e40 100644 --- a/infra/local/docker-compose.yml +++ b/infra/local/docker-compose.yml @@ -5,7 +5,7 @@ # # - Keycloak identity provider: http://localhost:58080/auth # - Lakekeeper iceberg catalog UI: http://localhost:58080/iceberg/ui. Log in with the credentials listed in keycloak/README -# - Trino query engine: http://localhost:58088 +# - Trino query engine: http://localhost:59088 # - Superset BI tool: http://localhost:58080/workspace/playground. Log in with the credentials defined in env-for-testing # # Login credentials @@ -97,7 +97,7 @@ services: start_period: 5s ports: - "58080:80" - - "58088:8088" # Trino won't allow itself to be deployed on a prefix path + - "59088:8088" # Trino won't allow itself to be deployed on a prefix path volumes: - "./traefik/traefik.yml:/etc/traefik/traefik.yml" - "./traefik/traefik-dynamic.yml:/etc/traefik/traefik-dynamic.yml" diff --git a/warehouses/accelerator/transform/accelerator/profiles.yml b/warehouses/accelerator/transform/accelerator/profiles.yml index f7587155..2570cdf6 100644 --- a/warehouses/accelerator/transform/accelerator/profiles.yml +++ b/warehouses/accelerator/transform/accelerator/profiles.yml @@ -9,7 +9,7 @@ trino_catalog: password: "{{ env_var('DBT_TRINO_PASSWORD', '') }}" http_scheme: "{{ env_var('DBT_TRINO_HTTP_SCHEME', 'http') }}" host: "{{ env_var('DBT_TRINO_HOST', 'analytics.localhost') }}" - port: "{{ env_var('DBT_TRINO_PORT', '58088') | int }}" + port: "{{ env_var('DBT_TRINO_PORT', '59088') | int }}" database: "{{ env_var('DBT_TRINO_CATALOG', 'accelerator') }}" schema: "{{ env_var('DBT_TRINO_CATALOG_SCHEMA_PREFIX', 'dev_') }}analytics" threads: "{{ env_var('DBT_TRINO_THREADS', '8') | int }}" From 38536e1af47fedbc65860f81c0e6909cdce6faa0 Mon Sep 17 00:00:00 2001 From: Martyn Gigg Date: Tue, 9 Dec 2025 17:00:41 +0000 Subject: [PATCH 3/3] Allow exception to just bubble to the surface --- .../dlt_destinations/pyiceberg/pyiceberg.py | 20 +++++++------------ .../pyiceberg/test_pyiceberg_pipeline.py | 4 +++- 2 files changed, 10 insertions(+), 14 deletions(-) diff --git a/elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py b/elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py index 581a94ae..7022d671 100644 --- a/elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py +++ b/elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py @@ -422,19 +422,13 @@ def write_to_table( # provide merge key columns if no identifier-field-ids exist to mark the primary key columns strategy = dlt_schema["x-merge-strategy"] # type:ignore if strategy == "upsert": - # requires the indentifier fields to have been defined - try: - table.upsert( - df=table_data, - join_cols=join_columns, - when_matched_update_all=True, - when_not_matched_insert_all=True, - case_sensitive=True, - ) - except ValueError as exc: - logger.info(f"----- Existing data ------\n{table.scan().to_arrow()}") - logger.info(f"----- New data ------\n{table_data}") - raise exc + table.upsert( + df=table_data, + join_cols=join_columns, + when_matched_update_all=True, + when_not_matched_insert_all=True, + case_sensitive=True, + ) else: raise DestinationTerminalException( f'Merge strategy "{strategy}" is not supported for Iceberg tables. ' diff --git a/elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py b/elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py index db714506..17c58f09 100644 --- a/elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py +++ b/elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py @@ -34,7 +34,9 @@ def pipeline_name(frame: FrameType | None): def resource_factory( - data: List[Dict[str, Any]] | None = None, primary_key: str | None = "id", merge_key=None + data: List[Dict[str, Any]] | None = None, + primary_key: str | None = "id", + merge_key: str | None = None, ): kwargs = {} if primary_key is not None: