diff --git a/elt-common/src/elt_common/cli.py b/elt-common/src/elt_common/cli.py index 5a28f546..bc53a6ed 100644 --- a/elt-common/src/elt_common/cli.py +++ b/elt-common/src/elt_common/cli.py @@ -78,7 +78,6 @@ def cli_main( :param data_generator: Callable returning a dlt.DltSource or dlt.DltResource :param dataset_name_suffix: Suffix part of full dataset name in the destination. The given string is prefixed with a standard string defined in constants.DATASET_NAME_PREFIX_SRCS - :param default_write_disposition: Default mode for dlt write_disposition defaults to "append" :param default_destination: Default destination, defaults to "filesystem" :param default_loader_file_format: Default dlt loader file format, defaults to "parquet" :param default_progress: Default progress reporter, defaults to NULL_COLLECTOR diff --git a/warehouses/accelerator/extract_load/accelerator_sharepoint/extract_and_load.py b/warehouses/accelerator/extract_load/accelerator_sharepoint/extract_and_load.py index 5f4dbf69..d21a243f 100644 --- a/warehouses/accelerator/extract_load/accelerator_sharepoint/extract_and_load.py +++ b/warehouses/accelerator/extract_load/accelerator_sharepoint/extract_and_load.py @@ -55,7 +55,8 @@ def equipment_downtime_records_archive() -> DltResource: columns={ "FaultDate": {"data_type": "text"}, "FaultTime": {"data_type": "text"}, - } + }, + write_disposition="replace", ) ) return reader @@ -69,8 +70,15 @@ def edr_equipment_mapping() -> DltResource: extract_content=True, ) return ( - files | read_excel(header=None, names=["equipment_name", "equipment_category"]) - ).with_name("edr_equipment_mapping") + ( + files + | read_excel(header=None, names=["equipment_name", "equipment_category"]) + ) + .with_name("edr_equipment_mapping") + .apply_hints( + write_disposition="replace", + ) + ) # We need to set the section in order for dlt to find our named secret. @@ -87,5 +95,4 @@ def resources(): default_destination="elt_common.dlt_destinations.pyiceberg", data_generator=resources(), dataset_name_suffix="accelerator_sharepoint", - default_write_disposition="replace", ) 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 a6cbe060..7bd52e7d 100644 --- a/warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py +++ b/warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py @@ -155,7 +155,9 @@ def rdm_data( - pendulum.Duration(seconds=ONE_DAY_SECS), ) reader = files | extract_content_and_read() - yield from reader + yield from reader.apply_hints( + write_disposition="merge", + ) cli_main( @@ -165,5 +167,4 @@ def rdm_data( rdm_data, partition=PartitionTrBuilder.year("date_time") ), dataset_name_suffix=PIPELINE_NAME, - default_write_disposition="merge", ) diff --git a/warehouses/accelerator/extract_load/statusdisplay/extract_and_load.py b/warehouses/accelerator/extract_load/statusdisplay/extract_and_load.py index 643797fb..874cd36f 100755 --- a/warehouses/accelerator/extract_load/statusdisplay/extract_and_load.py +++ b/warehouses/accelerator/extract_load/statusdisplay/extract_and_load.py @@ -25,6 +25,9 @@ def statusdisplay() -> DltSource: "base_url": dlt.config["sources.base_url"], }, "resources": dlt.config["sources.resources"], + "resource_defaults": { + "write_disposition": "replace", + }, }, ) @@ -36,5 +39,4 @@ def statusdisplay() -> DltSource: default_destination="elt_common.dlt_destinations.pyiceberg", data_generator=statusdisplay(), dataset_name_suffix="statusdisplay", - default_write_disposition="replace", )