fix(warehouses/accelerator): Electricity ingest: Skip files that cannot be processed. (#169)

### Summary

Catch errors loading the electricity consumption data. Users would
rather some occasional gaps (that can be backfilled) rather than data
loading that just stops.

Refs #166 




<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

## Release Notes

* **Bug Fixes**
* Enhanced error handling for merge operations with improved logging of
failed data loads.
* Improved exception handling during data ingestion to prevent silent
failures.

* **New Features**
* Extended merge configuration capabilities for incremental data
synchronisation.

* **Tests**
* Expanded test coverage for merge key configurations and error
scenarios.

<sub>✏️ Tip: You can customize this high-level summary in your review
settings.</sub>

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: martyngigg

elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py

@@ -411,11 +411,20 @@ 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,
+                    join_cols=join_columns,
                     when_matched_update_all=True,
                     when_not_matched_insert_all=True,
                     case_sensitive=True,

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 = ""
 

elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py

@@ -33,9 +33,19 @@ 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: str | None = 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 +178,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 +205,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 +222,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 +235,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,

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"

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",

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 }}"