Looking for best practices/performances working with high volume data in Fabric

I’m using DuckDB to read data from a OneLake Lakehouse and merge it into another table.

The dataset contains around 500M rows. When loaded entirely into memory, the process fails, so I implemented a batch-based iterative merge to avoid crashes.

I’m now looking for best practices and performance tuning guidance, as this pattern will be industrialized and used extensively.

Below is my current implementation, it works, but processing 5M-row batches takes ~45 minutes in a Fabric Python Notebook environment (8 vCores / 64 GB RAM):

import duckdb
import os
import time
import gc
import pyarrow as pa
from deltalake import DeltaTable, write_deltalake


BATCH_SIZE = 5_000_000 
TARGET_TABLE_NAME = "tbl_f_instr_price_500M"
TARGET_PATH = f"{TARGET_TABLES_BASE_PATH}/{TARGET_TABLE_NAME}"


sql_query = f"""
    SELECT 
        INSTR.ID_INSTRUMENT, 
        CCY.ID_CCY, 
        CCY.CD_CCY_ISO, 
        INSTR.CD_INSTRUMENT_SYMBOL,
        WK.*
    FROM delta_scan('{os.path.join(TABLES_PATH, 'fact_instrument_price_500M')}') WK
    LEFT OUTER JOIN delta_scan('{os.path.join(TABLES_PATH, 'dim_currency')}') CCY 
        ON WK.ID_CCY = CCY.ID_CCY
    LEFT OUTER JOIN delta_scan('{os.path.join(TABLES_PATH, 'dim_instrument')}') INSTR 
        ON WK.ID_INSTRUMENT = INSTR.ID_INSTRUMENT
"""


conn.execute(f"CREATE OR REPLACE VIEW WK_INSTR_PRICE_500M AS {sql_query}")


# Define the source query
clean_source_query = """
SELECT 
    ID_INSTRUMENT,
    ID_CCY,
    CD_CCY_ISO,
    ValuationDate AS DT_VALUATION,
    Value AS PR_UNIT
FROM WK_INSTR_PRICE_500M
"""


if not notebookutils.fs.exists(TARGET_PATH):
    print(f"Target table not found. Initializing with seed...")
    seed_arrow = conn.execute(f"{clean_source_query} LIMIT 1").fetch_arrow_table()
    write_deltalake(TARGET_PATH, seed_arrow, mode="overwrite")
    print("Initialization Complete.")


print(f"Starting Manual Batched Merge (Batch Size: {BATCH_SIZE:,})...")
start_time = time.time()


reader = conn.execute(clean_source_query).fetch_record_batch(rows_per_batch=BATCH_SIZE)


dt = DeltaTable(TARGET_PATH)
total_rows_processed = 0
batch_idx = 0


try:
    for batch in reader:
        batch_idx += 1
        
        source_chunk = pa.Table.from_batches([batch])
        row_count = source_chunk.num_rows
        
        print(f"Merging Batch {batch_idx} ({row_count:,} rows)...")


        (
            dt.merge(
                source=source_chunk,
                predicate="target.ID_INSTRUMENT = source.ID_INSTRUMENT AND target.DT_VALUATION = source.DT_VALUATION AND target.ID_CCY = source.ID_CCY",
                source_alias="source",
                target_alias="target"
            )
            .when_matched_update(
                updates={"PR_UNIT": "source.PR_UNIT"}
            )
            .when_not_matched_insert(
                updates={
                    "ID_INSTRUMENT": "source.ID_INSTRUMENT",
                    "DT_VALUATION": "source.DT_VALUATION",
                    "ID_CCY": "source.ID_CCY",
                    "CD_CCY_ISO": "source.CD_CCY_ISO",
                    "PR_UNIT": "source.PR_UNIT"
                }
            )
            .execute()
        )
        
        total_rows_processed += row_count
        
        del source_chunk
        del batch
        gc.collect()


except Exception as e:
    print(f"Error on batch {batch_idx}: {e}")
    raise e


end_time = time.time()
elapsed_time = end_time - start_time


print(f"Merge Complete.")
print(f"Total Batches: {batch_idx}")
print(f"Total Rows Processed: {total_rows_processed:,}")
print(f"Total time: {elapsed_time:.2f} seconds")