From c37c7cc8505e9df50eee3410b4a256f922bb4339 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 14:31:07 +1000 Subject: [PATCH 01/14] Fixed webapp plots --- vast_pipeline/plots.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/vast_pipeline/plots.py b/vast_pipeline/plots.py index b12845f8f..e16f5841a 100644 --- a/vast_pipeline/plots.py +++ b/vast_pipeline/plots.py @@ -84,6 +84,7 @@ def plot_lightcurve( # lightcurve required cols: taustart_ts, flux, flux_err_upper, flux_err_lower, forced lightcurve = pd.DataFrame(measurements_qs) + lightcurve['id'] = lightcurve['id'].astype(str) # remap method values to labels to make a better legend lightcurve["method"] = lightcurve.forced.map({True: "Forced", False: "Selavy"}) @@ -167,6 +168,13 @@ def plot_lightcurve( ) .reset_index() ) + + candidate_measurement_pairs_df = candidate_measurement_pairs_df.astype( + {'measurement_a_id': 'str', + 'measurement_b_id': 'str', + } + ) + g = nx.Graph() for _row in candidate_measurement_pairs_df.itertuples(index=False): g.add_edge(_row.measurement_a_id, _row.measurement_b_id) From 2da72d5cf1ae609f20df20432a9b6849451f3f42 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 14:32:44 +1000 Subject: [PATCH 02/14] Updated changelog --- CHANGELOG.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6abcf4a4b..8e1aa87cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,6 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), #### Added - - V2: Migrate pipeline to used a Dask.distributed.LocalCluster throughout [#816](https://github.com/askap-vast/vast-pipeline/pull/816) - V2: Add Dask.distributed support - V2: Use `django-postgres-copy` for database uploads [#803](https://github.com/askap-vast/vast-pipeline/pull/803) @@ -35,6 +34,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), #### Fixed +- V2: Fixed problem with webapp plots related to UUIDs [#844](https://github.com/askap-vast/vast-pipeline/pull/844) - V2: Pair metrics working with V2 dask LocalCluster changes [#817](https://github.com/askap-vast/vast-pipeline/pull/817) - V2: Fixed missing JS9 overlays on source webpage [#809](https://github.com/askap-vast/vast-pipeline/pull/809) - V2: Fix bug when deleting source tags from database which are referenced by multiple sources [#803](https://github.com/askap-vast/vast-pipeline/pull/803) @@ -58,6 +58,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), #### List of PRs +- [#844](https://github.com/askap-vast/vast-pipeline/pull/844): fix: V2: Fixed problem with webapp plots related to UUIDs - [#833](https://github.com/askap-vast/vast-pipeline/pull/833): feat: V2: Limit associations upload to using num_io_workers - [#829](https://github.com/askap-vast/vast-pipeline/pull/829): feat: V2: Allow user specification of dask dashboard paramters and add some further logging to dask setup - [#817](https://github.com/askap-vast/vast-pipeline/pull/817): fix: V2: Updates to pairs calculation to make it work with Dask `LocalCluster`. From c73b24feaf05c62eb53200bf93c5a60858303367 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 14:43:54 +1000 Subject: [PATCH 03/14] measurement UUID tweaks --- vast_pipeline/image/main.py | 7 +++--- ...ter_association_id_alter_measurement_id.py | 24 +++++++++++++++++++ vast_pipeline/models.py | 5 ++-- vast_pipeline/pipeline/forced_extraction.py | 10 ++++---- vast_pipeline/pipeline/loading.py | 3 ++- vast_pipeline/serializers.py | 2 +- vast_pipeline/urls.py | 5 ++++ 7 files changed, 42 insertions(+), 14 deletions(-) create mode 100644 vast_pipeline/migrations/0003_alter_association_id_alter_measurement_id.py diff --git a/vast_pipeline/image/main.py b/vast_pipeline/image/main.py index 0a014636c..c6219f6b5 100644 --- a/vast_pipeline/image/main.py +++ b/vast_pipeline/image/main.py @@ -6,6 +6,7 @@ import logging import numpy as np import pandas as pd +import uuid from django.conf import settings from astropy.io import fits @@ -18,9 +19,7 @@ from vast_pipeline import models from vast_pipeline.survey.translators import tr_selavy -from vast_pipeline.utils.utils import ( - generate_shortuuid, UUID_LEN_MEAS, -) + from vast_pipeline.image.utils import get_fits_header @@ -341,7 +340,7 @@ def read_selavy(self, dj_image: models.Image) -> pd.DataFrame: df[key["name"]] = df[key["name"]].astype(key["dtype"]) # Add id column - df["id"] = df.apply(lambda _: generate_shortuuid(UUID_LEN_MEAS), axis=1) + df["id"] = df.apply(lambda _: str(uuid.uuid4()), axis=1) # do checks and fill in missing field for uploading sources # in DB (see fields in models.py -> Source model) diff --git a/vast_pipeline/migrations/0003_alter_association_id_alter_measurement_id.py b/vast_pipeline/migrations/0003_alter_association_id_alter_measurement_id.py new file mode 100644 index 000000000..453df72c4 --- /dev/null +++ b/vast_pipeline/migrations/0003_alter_association_id_alter_measurement_id.py @@ -0,0 +1,24 @@ +# Generated by Django 5.1.6 on 2025-03-30 23:19 + +import uuid +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('vast_pipeline', '0002_q3c'), + ] + + operations = [ + migrations.AlterField( + model_name='association', + name='id', + field=models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False), + ), + migrations.AlterField( + model_name='measurement', + name='id', + field=models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False), + ), + ] diff --git a/vast_pipeline/models.py b/vast_pipeline/models.py index 8666970c0..e99ea8cdd 100644 --- a/vast_pipeline/models.py +++ b/vast_pipeline/models.py @@ -1,4 +1,5 @@ import numpy as np +import uuid from dataclasses import dataclass from itertools import combinations @@ -417,7 +418,7 @@ class Measurement(CommentableModel): Essentially a source single measurement in time. """ - id = ShortUUIDField(primary_key=True, editable=False, length=UUID_LEN_MEAS, alphabet=UUID_ALPHABET) + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) image = models.ForeignKey( Image, null=True, on_delete=models.CASCADE, to_field="id" ) # first image seen in @@ -713,7 +714,7 @@ class Association(models.Model): some parameters """ - id = ShortUUIDField(primary_key=True, editable=False, length=UUID_LEN_MEAS, alphabet=UUID_ALPHABET) + id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) source = models.ForeignKey(Source, on_delete=models.CASCADE, to_field="id") meas = models.ForeignKey(Measurement, on_delete=models.CASCADE, to_field="id") diff --git a/vast_pipeline/pipeline/forced_extraction.py b/vast_pipeline/pipeline/forced_extraction.py index 3c3779c3e..809fab5c0 100644 --- a/vast_pipeline/pipeline/forced_extraction.py +++ b/vast_pipeline/pipeline/forced_extraction.py @@ -1,6 +1,8 @@ import os import logging import datetime +import uuid + import numpy as np import pandas as pd import dask.dataframe as dd @@ -20,11 +22,7 @@ from vast_pipeline.pipeline.loading import copy_upload_measurements from forced_phot import ForcedPhot -from ..utils.utils import ( - StopWatch, - generate_shortuuid, - UUID_LEN_MEAS -) +from ..utils.utils import StopWatch from vast_pipeline.image.utils import open_fits # NOTE: We check here to see if we're in a testing environment. @@ -486,7 +484,7 @@ def _update_forced_measurements(df: pd.DataFrame) -> pd.DataFrame: The forced extraction dataframe updated with defaults. """ df["name"] = df["name"] + f"_f_{p_run_id}" - df["id"] = df.apply(lambda _: generate_shortuuid(UUID_LEN_MEAS), axis=1) + df["id"] = df.apply(lambda _: str(uuid.uuid4()), axis=1) default_pos_err = settings.POS_DEFAULT_MIN_ERROR / 3600.0 df["ra_err"] = default_pos_err df["dec_err"] = default_pos_err diff --git a/vast_pipeline/pipeline/loading.py b/vast_pipeline/pipeline/loading.py index c7651762a..8a0aa3a84 100644 --- a/vast_pipeline/pipeline/loading.py +++ b/vast_pipeline/pipeline/loading.py @@ -9,6 +9,7 @@ from itertools import islice from django.db import transaction, connection, models from contextlib import closing +from uuid import uuid4 from vast_pipeline.image.main import SelavyImage from vast_pipeline.pipeline.model_generator import ( @@ -432,7 +433,7 @@ def copy_upload_associations( } def upload(df, Association, mapping, batch_size): - df["db_id"] = df.apply(lambda _: generate_shortuuid(UUID_LEN_MEAS), axis=1) + df["db_id"] = df.apply(lambda _: str(uuid4()), axis=1) copy_upload_model(df, Association, mapping=mapping, batch_size=batch_size) associations_df = associations_df[columns_to_upload].map_partitions(upload, diff --git a/vast_pipeline/serializers.py b/vast_pipeline/serializers.py index cc889cf8f..866ab5425 100644 --- a/vast_pipeline/serializers.py +++ b/vast_pipeline/serializers.py @@ -56,7 +56,7 @@ class Meta: class MeasurementSerializer(serializers.ModelSerializer): - id = serializers.CharField(read_only=True) + id = serializers.UUIDField(read_only=True) frequency = serializers.SerializerMethodField(read_only=True) def get_frequency(self, obj): diff --git a/vast_pipeline/urls.py b/vast_pipeline/urls.py index 9581e36e0..02e986b6f 100644 --- a/vast_pipeline/urls.py +++ b/vast_pipeline/urls.py @@ -38,6 +38,11 @@ name="image_detail", ), path("measurements/", views.MeasurementIndex, name="measurement_index"), + path( + "measurements//", + views.MeasurementDetail, + name="measurement_detail" + ), re_path( fr"^measurements/(?P[\w]{{{UUID_LEN_MEAS}}})(?:/(?P[\w]+))?/$", views.MeasurementDetail, From 29c928b2671b36b9b19ca95ef0bdd531f7087499 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 14:46:27 +1000 Subject: [PATCH 04/14] PEP8 --- vast_pipeline/image/main.py | 2 +- vast_pipeline/pipeline/loading.py | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/vast_pipeline/image/main.py b/vast_pipeline/image/main.py index c6219f6b5..1ec11a261 100644 --- a/vast_pipeline/image/main.py +++ b/vast_pipeline/image/main.py @@ -15,7 +15,7 @@ from astropy.wcs.utils import proj_plane_pixel_scales from typing import Dict -from .utils import calc_condon_flux_errors, open_fits +from .utils import calc_condon_flux_errors from vast_pipeline import models from vast_pipeline.survey.translators import tr_selavy diff --git a/vast_pipeline/pipeline/loading.py b/vast_pipeline/pipeline/loading.py index 8a0aa3a84..0d5ef6e19 100644 --- a/vast_pipeline/pipeline/loading.py +++ b/vast_pipeline/pipeline/loading.py @@ -37,7 +37,6 @@ deg2hms, deg2dms, generate_shortuuid, - UUID_LEN_MEAS, UUID_LEN_SOURCE ) From ee4b8da46f5ca1a32401980d79468b469ee26a3a Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 15:02:02 +1000 Subject: [PATCH 05/14] Changelog updates --- CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 73a923cee..740742e99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), #### Changed +- V2: Change measurement and association DB indices to UUIDs [#844](https://github.com/askap-vast/vast-pipeline/pull/844) - V2: Limit associations upload to using num_io_workers [#833](https://github.com/askap-vast/vast-pipeline/pull/833) - V2: Allow user specification of dask dashboard paramters and add some further logging to dask setup [#829](https://github.com/askap-vast/vast-pipeline/pull/829) - V2: Replace all source, measurement, run etc. IDs with UUID indices. [#803](https://github.com/askap-vast/vast-pipeline/pull/803) @@ -35,7 +36,6 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), #### Fixed -- V2: Fixed problem with webapp plots related to UUIDs [#844](https://github.com/askap-vast/vast-pipeline/pull/844) - V2: Pair metrics working with V2 dask LocalCluster changes [#817](https://github.com/askap-vast/vast-pipeline/pull/817) - V2: Fixed missing JS9 overlays on source webpage [#809](https://github.com/askap-vast/vast-pipeline/pull/809) - V2: Fix bug when deleting source tags from database which are referenced by multiple sources [#803](https://github.com/askap-vast/vast-pipeline/pull/803) @@ -59,7 +59,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), #### List of PRs -- [#844](https://github.com/askap-vast/vast-pipeline/pull/844): fix: V2: Fixed problem with webapp plots related to UUIDs +- [#844](https://github.com/askap-vast/vast-pipeline/pull/844): fix: V2: Change measurements and association DB indices to UUIDs - [#843](https://github.com/askap-vast/vast-pipeline/pull/843): feat: V2: Enable specification of Dask worker memory limits - [#833](https://github.com/askap-vast/vast-pipeline/pull/833): feat: V2: Limit associations upload to using num_io_workers - [#829](https://github.com/askap-vast/vast-pipeline/pull/829): feat: V2: Allow user specification of dask dashboard paramters and add some further logging to dask setup From 535198004b5806f08f09968507660caa374ff37c Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 15:21:51 +1000 Subject: [PATCH 06/14] Fix memory leak in image upload --- vast_pipeline/pipeline/loading.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/vast_pipeline/pipeline/loading.py b/vast_pipeline/pipeline/loading.py index c7651762a..f2438b2ed 100644 --- a/vast_pipeline/pipeline/loading.py +++ b/vast_pipeline/pipeline/loading.py @@ -1,5 +1,7 @@ import os import logging +import gc + import numpy as np import pandas as pd import dask.dataframe as dd @@ -196,7 +198,8 @@ def make_upload_images( os.makedirs(base_folder) measurements.to_parquet(img.measurements_path, index=False) - del measurements, image, band, img + del measurements, image, band + gc.collect() logger.info("Total images upload/loading time: %.2f seconds", timer.reset_init()) From c7a158123da7a2df54f44ccc06c2a5587d8e2dd1 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 15:42:04 +1000 Subject: [PATCH 07/14] Fixed forced extraction memory leak --- vast_pipeline/pipeline/forced_extraction.py | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/vast_pipeline/pipeline/forced_extraction.py b/vast_pipeline/pipeline/forced_extraction.py index 3c3779c3e..cf2a6c6d5 100644 --- a/vast_pipeline/pipeline/forced_extraction.py +++ b/vast_pipeline/pipeline/forced_extraction.py @@ -682,7 +682,7 @@ def forced_extraction( # Explode out the img_diff column. extr_df = extr_df.explode("img_diff").reset_index() - total_to_extract = extr_df.shape[0] + total_to_extract = extr_df.shape[0].compute() if add_mode: # If we are adding images to the run we assume that monitoring was @@ -693,7 +693,7 @@ def forced_extraction( # images. # 3. A new relation has been created and they need the forced # measuremnts filled in (actually covered by 2.) - total_to_extract = extr_df.shape[0].compute() + logger.info("Running extraction in add mode...") extr_df = dd.concat( [ extr_df[~extr_df["img_diff"].isin(done_images_df["name"])], @@ -715,12 +715,14 @@ def forced_extraction( min_sigma, edge_buffer, cluster_threshold, allow_nan, add_mode, p_run.path, io_workers ) + logger.info("Completed parallel extraction step.") # Dask needs type metadata for map_partitions sources_meta = dd.utils.make_meta(sources_df).drop(['epoch', 'interim_ns', 'interim_ew'], axis=1) # Get expected database measurements schema columns = read_schema(images_df.iloc[0]["measurements_path"]).names + logger.debug("Building save and upload...") extr_df = extr_df.map_partitions(save_and_upload_forced_df, p_run_path=p_run.path, p_run_id=p_run.id, @@ -744,18 +746,27 @@ def forced_extraction( else: extr_df["epoch"] = sources_df['epoch'].compute().iloc[0] + logger.info("Persisting extr_df...") + extr_df = extr_df.persist() + wait(extr_df) + logger.info("Persisted extr_df.") + sources_df = dd.concat( - [sources_df, extr_df] + [sources_df, extr_df], + interleave_partitions=True ) # Wait for the forced extraction step to complete # NOTE: Ideally we would have some optimised way of sorting sources_df # by source id at this point to avoid needing to `set_index` on it # during the finalise step. + logger.info("Persisting sources_df...") sources_df = sources_df.persist() wait(sources_df) + logger.info("Persisted sources_df") del extr_df + gc.collect() # get the number of forced extractions for the run forced_parquets = glob(os.path.join(p_run.path, "forced_measurements*.parquet")) From 9c8454c3be063e3d740327b037132939d6692560 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 16:18:45 +1000 Subject: [PATCH 08/14] Fix imports --- vast_pipeline/pipeline/forced_extraction.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/vast_pipeline/pipeline/forced_extraction.py b/vast_pipeline/pipeline/forced_extraction.py index cf2a6c6d5..c703fb3a0 100644 --- a/vast_pipeline/pipeline/forced_extraction.py +++ b/vast_pipeline/pipeline/forced_extraction.py @@ -1,6 +1,8 @@ import os import logging import datetime +import gc + import numpy as np import pandas as pd import dask.dataframe as dd From 107b29bbd72eb70ed29460c2b77fcb6acba92ed2 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 17:10:55 +1000 Subject: [PATCH 09/14] Fix duplicate source ID issue --- vast_pipeline/pipeline/finalise.py | 11 +++++++++-- vast_pipeline/utils/utils.py | 16 +++++++++------- 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/vast_pipeline/pipeline/finalise.py b/vast_pipeline/pipeline/finalise.py index afec07aea..c83ea10ae 100644 --- a/vast_pipeline/pipeline/finalise.py +++ b/vast_pipeline/pipeline/finalise.py @@ -96,8 +96,15 @@ def final_operations( logger.info("Calculating statistics for sources...") log_total_memory_usage() - sources_df = sources_df.set_index("source") \ - .repartition(partition_size=f"{upload_chunk_size_mb}MB") + npartitions = calculate_n_partitions( + sources_df, + partition_size_mb=upload_chunk_size_mb + ) + + with dc.set({"dataframe.shuffle.method": "p2p"}): + sources_df = sources_df.set_index("source") \ + .shuffle(npartitions=npartitions, on_index=True) + srcs_df = parallel_groupby(sources_df) mem_usage = get_df_memory_usage(srcs_df) diff --git a/vast_pipeline/utils/utils.py b/vast_pipeline/utils/utils.py index f8b605fad..c9fef6f4f 100644 --- a/vast_pipeline/utils/utils.py +++ b/vast_pipeline/utils/utils.py @@ -393,8 +393,8 @@ def timeStamped(fname, fmt="%Y-%m-%d-%H-%M-%S_{fname}"): def calculate_n_partitions( - df: pd.DataFrame, - n_cpu: int, + df: Union[pd.DataFrame, dd.DataFrame], + n_cpu: Optional[int] = None, partition_size_mb: Optional[int] = 15 ) -> int: """ @@ -427,15 +427,17 @@ def calculate_n_partitions( n_partitions = int(np.ceil(mem_usage_mb / partition_size_mb)) # n_partitions should be >= n_cpu for optimal parallel processing - if n_partitions < n_cpu: - n_partitions = n_cpu + if n_cpu is not None: + if n_partitions < n_cpu: + n_partitions = n_cpu - partition_size_mb = int(np.ceil(mem_usage_mb / n_partitions)) + partition_size_mb = mem_usage_mb / n_partitions logger.debug( - "Using %d partitions of %dMB", + "Using %d partitions of %.1fMB", n_partitions, - partition_size_mb) + partition_size_mb + ) return n_partitions From f77bbd63ce77cbc4fa4d66638ecb2c03a3d9f641 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 18:07:26 +1000 Subject: [PATCH 10/14] Tweaks to get this running --- vast_pipeline/pipeline/finalise.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/vast_pipeline/pipeline/finalise.py b/vast_pipeline/pipeline/finalise.py index c83ea10ae..61809b8c0 100644 --- a/vast_pipeline/pipeline/finalise.py +++ b/vast_pipeline/pipeline/finalise.py @@ -1,18 +1,24 @@ import os import logging import warnings + import pandas as pd import pyarrow as pa import dask.dataframe as dd +import dask.config as dc from astropy import units as u from astropy.coordinates import SkyCoord from django.conf import settings from typing import List, Dict, Tuple +from dask.distributed import wait from vast_pipeline.models import Run from vast_pipeline.utils.utils import ( - StopWatch, optimise_numeric, delete_file_or_dir + StopWatch, + optimise_numeric, + delete_file_or_dir, + calculate_n_partitions ) from vast_pipeline.pipeline.loading import ( update_sources, From fe3d287c1ad90d3be8bf46e29d5fdb0fe266f7b0 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 18:23:15 +1000 Subject: [PATCH 11/14] Updated changelog --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 740742e99..68d8a148a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), #### Fixed +- V2: Fixed duplicate source ID error by switching from sources_df.repartition() to sources_df.shuffle() in pipeline.finalise.final_operations [#845](https://github.com/askap-vast/vast-pipeline/pull/845) +- V2: Tentative fix for image upload memory leak via garbage collect [#845](https://github.com/askap-vast/vast-pipeline/pull/845) +- V2: Fixed dd.concat memory blow-up in pipeline.forced_extraction.forced_extraction by persisting both dataframes prior [#845](https://github.com/askap-vast/vast-pipeline/pull/845) - V2: Pair metrics working with V2 dask LocalCluster changes [#817](https://github.com/askap-vast/vast-pipeline/pull/817) - V2: Fixed missing JS9 overlays on source webpage [#809](https://github.com/askap-vast/vast-pipeline/pull/809) - V2: Fix bug when deleting source tags from database which are referenced by multiple sources [#803](https://github.com/askap-vast/vast-pipeline/pull/803) @@ -59,6 +62,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), #### List of PRs +- [#845](https://github.com/askap-vast/vast-pipeline/pull/845): fix: V2: Fix memory leaks and duplicate source ID issue - [#844](https://github.com/askap-vast/vast-pipeline/pull/844): fix: V2: Change measurements and association DB indices to UUIDs - [#843](https://github.com/askap-vast/vast-pipeline/pull/843): feat: V2: Enable specification of Dask worker memory limits - [#833](https://github.com/askap-vast/vast-pipeline/pull/833): feat: V2: Limit associations upload to using num_io_workers From d9fa83a2b42cbf2efbe0075cd940ea375af61524 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Tue, 20 May 2025 18:24:02 +1000 Subject: [PATCH 12/14] PEP8 --- vast_pipeline/pipeline/finalise.py | 1 - 1 file changed, 1 deletion(-) diff --git a/vast_pipeline/pipeline/finalise.py b/vast_pipeline/pipeline/finalise.py index 61809b8c0..34a8b49a9 100644 --- a/vast_pipeline/pipeline/finalise.py +++ b/vast_pipeline/pipeline/finalise.py @@ -11,7 +11,6 @@ from astropy.coordinates import SkyCoord from django.conf import settings from typing import List, Dict, Tuple -from dask.distributed import wait from vast_pipeline.models import Run from vast_pipeline.utils.utils import ( From b69c345987f7ce9d952bb0c057ed1cd81d969541 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Wed, 21 May 2025 10:41:12 +1000 Subject: [PATCH 13/14] Tweak association upload and persist measuremnet pairs df --- vast_pipeline/pipeline/finalise.py | 6 +++--- vast_pipeline/pipeline/loading.py | 29 +++++++++++------------------ 2 files changed, 14 insertions(+), 21 deletions(-) diff --git a/vast_pipeline/pipeline/finalise.py b/vast_pipeline/pipeline/finalise.py index 34a8b49a9..59322a14e 100644 --- a/vast_pipeline/pipeline/finalise.py +++ b/vast_pipeline/pipeline/finalise.py @@ -343,8 +343,8 @@ def final_operations( # upload associations into DB if not __TESTING__: - assoc_df = associations_df_upload.loc[:, ["id", "source", "d2d", "dr"]] - copy_upload_associations(assoc_df, io_workers) + associations_df_upload = associations_df_upload.loc[:, ["id", "source", "d2d", "dr"]] + copy_upload_associations(associations_df_upload) # write associations to parquet file associations_df[['source', 'id', 'd2d', 'dr']] \ @@ -359,7 +359,7 @@ def final_operations( timer.reset() # ingest to dask data frames srcs_df.index.name = "source_id" - srcs_df = dd.from_pandas(srcs_df, npartitions=n_partitions) + srcs_df = dd.from_pandas(srcs_df, npartitions=n_partitions).persist() columns = ['id_a', 'id_b', 'flux_int_a', 'flux_int_err_a', 'flux_peak_a', 'flux_peak_err_a', 'image_name_a', 'flux_int_b', 'flux_int_err_b', 'flux_peak_b', 'flux_peak_err_b', 'image_name_b', 'vs_peak', 'vs_int', diff --git a/vast_pipeline/pipeline/loading.py b/vast_pipeline/pipeline/loading.py index cd9b6d9e5..437160ba8 100644 --- a/vast_pipeline/pipeline/loading.py +++ b/vast_pipeline/pipeline/loading.py @@ -406,19 +406,15 @@ def make_upload_related_sources(related_df: pd.DataFrame) -> None: def copy_upload_associations( associations_df: dd.DataFrame, - io_workers: List[str], batch_size: int = 10_000, ) -> None: """Upload associations using django-postgres-copy in-memory csv method. Args: associations_df: The associations dataframe to upload. - io_workers: - List of dask worker addresses to use for the compute. - This is likely the output of `DaskManager.get_n_random_workers()`. batch_size: The batch size. Defaults to 10_000. """ - logger.info("Upload associations...") + logger.info("Uploading associations in batches of %d", batch_size) columns_to_upload = ["source"] for fld in Association._meta.get_fields(): if getattr(fld, "attname", None) and fld.attname in associations_df.columns: @@ -433,19 +429,16 @@ def copy_upload_associations( "d2d": "d2d", "dr": "dr" } - - def upload(df, Association, mapping, batch_size): - df["db_id"] = df.apply(lambda _: str(uuid4()), axis=1) - copy_upload_model(df, Association, mapping=mapping, batch_size=batch_size) - - associations_df = associations_df[columns_to_upload].map_partitions(upload, - Association, - mapping, - batch_size, - enforce_metadata=False, - meta={}) - - associations_df.compute(workers=io_workers) + timer = StopWatch() + associations_df = associations_df.compute() + logger.debug("Time to compute associations_df: %.1f s", timer.reset()) + + associations_df["db_id"] = associations_df.apply(lambda _: str(uuid4()), axis=1) + logger.debug("Time to add db_id: %.1f s", timer.reset()) + + copy_upload_model(associations_df, Association, mapping=mapping, batch_size=batch_size) + logger.debug("Time to upload associations: %.1f s", timer.reset()) + logger.info("Associations upload complete") def make_upload_associations(associations_df: pd.DataFrame) -> None: From 49c4a5bfd8169c53cfb6c7d2b6c86f6d771c9698 Mon Sep 17 00:00:00 2001 From: Dougal Dobie Date: Wed, 21 May 2025 10:44:21 +1000 Subject: [PATCH 14/14] Updated changelog --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 68d8a148a..457def5da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), #### Changed +- V2: Compute associations prior to upload, and then upload in serial [#846](https://github.com/askap-vast/vast-pipeline/pull/846) +- V2: Persist srcs_df prior to measurement pairs df output file computation [#846](https://github.com/askap-vast/vast-pipeline/pull/846) - V2: Change measurement and association DB indices to UUIDs [#844](https://github.com/askap-vast/vast-pipeline/pull/844) - V2: Limit associations upload to using num_io_workers [#833](https://github.com/askap-vast/vast-pipeline/pull/833) - V2: Allow user specification of dask dashboard paramters and add some further logging to dask setup [#829](https://github.com/askap-vast/vast-pipeline/pull/829) @@ -62,6 +64,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), #### List of PRs +- [#846](https://github.com/askap-vast/vast-pipeline/pull/846): fix: V2: Compute associations prior to upload and persist srcs_df in final pairs calculation - [#845](https://github.com/askap-vast/vast-pipeline/pull/845): fix: V2: Fix memory leaks and duplicate source ID issue - [#844](https://github.com/askap-vast/vast-pipeline/pull/844): fix: V2: Change measurements and association DB indices to UUIDs - [#843](https://github.com/askap-vast/vast-pipeline/pull/843): feat: V2: Enable specification of Dask worker memory limits