Repository navigation
orchestration: create a staging cluster when the steps before its first steps end - #243
Closed
project-defiant wants to merge 17 commits into
Closed
project-defiant wants to merge 17 commits into
project-defiant wants to merge 17 commits into
Conversation
… it (#221) * orchestration: add StagingDagConfig, a validated config and builder for staging DAGs The config of a staging DAG is the file dags/<dag_id>/config.yaml, with the environments, the steps and the clusters of the DAG. The output of a run is a folder of the data bucket named after a timestamp, <bucket>/<env>/<timestamp>/, set in the config for each run of the DAG, as the run_name of the unified pipeline, with the rendered config in config/, the setup files (such as cluster init scripts) in bin/, and the datasets in output/. - models/staging_dag: the config schema. StagingEnvironments is the model of env and environments, validated once, with the input/data bucket rules (a run folder named after a timestamp, each test input root being the prod root or a dated static slice). StagingDagConfigModel is the model of the rest of the rendered config - dags/config/staging_pipeline.py: StagingDagConfig, the loader. The environments are validated on the raw config, the whole config is rendered as text with AppConfig ({{input.<name>}}, {{output}} and the scratchpad of the selected environment), the rendered text is checked for placeholders left, and the rest of the config is validated - models/staging_dag/step.py: a step is {kind, depends_on, definition}. The kind selects the model of the definition, and build() maps the definition to the task group of the step: a PySpark job on a Dataproc cluster, the Batch index, jobs and collect, or a container run on a VM and the deletion of the VM - utils/staging.py: build_staging_dag, the task builder: the config upload, the setup file copies into bin/, a create task per cluster, the tasks of each step, and a delete task per cluster. The clusters and the VMs of a run are named <name>-{{ run_id | strhash }}, as in the unified pipeline Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * orchestration: run the gwas catalog top hits DAG from a StagingDagConfig The top hits DAG moves to dags/gwas_catalog_top_hits/, with a config.yaml read with StagingDagConfig, on gentropy v4.0.0-dev.4 in prod. - the output of the run is the new folder gs://gwas_catalog_top_hits/prod/20261028T234317/, set in the config, instead of the folder of the 2026-08-28 release - the config of the run is recorded in config/. The gentropy installation on the cluster comes from the init script and CLI of the tag, uploaded to bin/, instead of the shared gs://genetics_etl_python_playground/initialisation/ scripts - the pics step has the v4 study_index_path argument, and its outputs are credible_set and credible_set_variant Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * orchestration: forbid unknown keys in the Batch models The Batch specs ignored unknown keys, so a typo in a Batch config, such as `paralelism` or `max_tasks_count`, was dropped and the default used. Every model of models/batch now forbids unknown keys, as the staging models do. The Batch specs of the existing DAG configs validate with the stricter models, and every DAG loads. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * orchestration: document the gwas catalog top hits run 20261028T234317 - the README describes the run prod/20261028T234317 on gentropy v4.0.0-dev.4: the steps ingestion, ld_annotation and pics, and the outputs study, study_locus, ld_annotated_study_locus, credible_set and credible_set_variant, with their schemas as read from the run - the DAG svg is generated from the StagingDagConfig DAG - upload-gwas-catalog-top-hits-readme takes the run folder as BUCKET and is no longer part of update-bucket-docs: each run folder has its own docs, uploaded once per run Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
![opentargets-gitbot[bot]](https://github.githubassets.com/images/gravatars/gravatar-user-420.png){kind=avatar}
…a dated prod run folder (#226) The 2026-08-28 summary statistics PICS run now lives in gs://gwas_catalog_sumstats_pics/prod/20260828T115527/ with its outputs under output/, its DAG config under config/ and its docs under docs/. - gwas_catalog_sumstats_pics.yaml: output_path points at the run's output/ folder - pis/config.yaml: GCSS PICS study index and credible sets read from the run folder - docs: split the sumstats PICS docs into gwas_catalog_sumstats_pics/README.md, with the inputs, gentropy links pinned to v3.4.0-rc.3 and the output schemas; regenerate the DAG svg - Makefile: upload-gwas-catalog-sumstats-pics-readme uploads only the sumstats PICS docs to a run folder given by BUCKET; the GWAS Catalog upload no longer copies the full docs to the sumstats PICS bucket Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
orchestration to 26.12.0rc19, pis to 26.12.0rc8
…nfig (#227) * orchestration: add the finngen multiome ingestion DAG on StagingDagConfig Run the gentropy step finngen_multiome_ingestion (opentargets/gentropy#1307) on gentropy v4.0.0-dev.5, from a StagingDagConfig in dags/finngen_multiome_ingestion/. - input: gs://finngen-public-data-multiome/eQTL/ (finemapping and cis_nominal) - output: gs://finngen_data/multiome/prod/20261029T123632/, with study, credible_set and credible_set_variant under output/ - one step on one Dataproc cluster, with the shape of the top hits cluster - docs: a FinnGen multiome section and a changelog entry in the finngen_data README Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * orchestration: finngen multiome DAG settings of the run 20261029T125742 - output: the run folder gs://finngen_data/multiome/prod/20261029T125742/ - cluster: spark.jars.packages=org.seqdoop:hadoop-bam:7.10.0. The step reads the inputs with the BGZFEnhancedGzipCodec; gentropy adds the package in the session config, too late for spark-submit on Dataproc, and the job failed with "Compression codec ... BGZFEnhancedGzipCodec not found" - cluster: 16 workers instead of 4. The inputs are about 340 GB of gzipped TSVs; with 64 cores 62 executors were pending Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
* orchestration: add the bigbrain ingestion DAG on StagingDagConfig Run the gentropy step bigbrain_finemapping_ingestion (opentargets/gentropy#1321) on gentropy v4.0.0-dev.5, from a StagingDagConfig in dags/bigbrain_ingestion/. - download: a VM step copies the published eQTL and sQTL fine-mapping and the sQTL top_assoc file from Zenodo into source/ of the run - eqtl and sqtl: one Dataproc step per QTL type; sqtl resolves junctions with the target index of the platform run do/platform-2610-2 - output: gs://bigbrain_data/prod/20261029T124825/, with study, credible_set and credible_set_variant under output/eqtl/ and output/sqtl/ - docs: docs/datasources/bigbrain_data/README.md Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * orchestration: bigbrain DAG output of the run 20261029T125742 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * orchestration: describe the bigbrain outputs of the run 20261029T125742 The same README goes to docs/ of the run folder. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
…od run folder (#229) * orchestration, pis: move the finngen r12 gwas release into a dated prod run folder The 2026-05-26 FinnGen R12 run now lives in gs://finngen_data/gwas/prod/20260526T131014/ with its outputs under output/, its DAG config under config/ and its docs under docs/. The run folder is named after the write time of the first output (study_index). The copy matches gs://finngen_data/r12/ object by object (name, size, crc32c). - finngen_ingestion.yaml: output_path points at the run's output/ folder - pis/config.yaml: FinnGen study index and credible sets read from the run folder - docs: split the FinnGen GWAS docs into finngen_data/finngen_gwas/README.md, with the inputs, gentropy links pinned to v3.2.0 and the output schemas; regenerate the DAG svg; finngen_data/README.md becomes the bucket index - Makefile: upload-finngen-gwas-readme uploads the FinnGen GWAS docs to a run folder given by BUCKET; upload-finngen-bucket-readme is removed with the top-level gs://finngen_data/docs/ Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * orchestration: add the finngen multiome docs to its run folder - docs: finngen_data/finngen_multiome/README.md, with the inputs, the filters of the ingestion step, gentropy links pinned to v4.0.0-dev.5 and the output schemas of study, credible_set and credible_set_variant; the DAG svg; the bucket index links to it - Makefile: upload-finngen-multiome-readme uploads the docs to a run folder given by BUCKET; build-dag-svgs renders the multiome DAG Uploaded to gs://finngen_data/multiome/prod/20261029T125742/docs/. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * orchestration: address the review of the finngen freeze - finngen_gwas README: region of chromosome X is chrX, except in 87 credible sets of FINNGEN_R12_HEIGHT_IRN, which FinnGen labels chr23 - Makefile: the finngen gwas and multiome upload targets are .PHONY Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
orchestration to 26.12.0rc22, pis to 26.12.0rc9
* pts: variant index in polars (POC) * pts: variant annotation parts tolerate column drift between parts annotation_parts_by_chromosome scanned every part with the first part's schema, so a later part with a column the first lacked raised a SchemaError before any annotation was read. It now scans only `chromosome`, ignoring other columns, as read_variant_annotations already does for the same parts. * pts: variant step lists parts through a shared helper and skips FoldX keys without FoldX list_parts in utils/dataset.py now holds the directory-or-file part listing that scan_dataset, list_vep_parts and annotation_parts_by_chromosome each repeated. The variant step collects the amino-acid keys of the parsed index only when an amino-acid annotation source is configured. * pts: variant step becomes the transformers/variant package variant.py and its private variant_utils/ helpers move into one transformers/variant/ package, as transformers/target/ is laid out, so the step's own modules are no longer confused with the shared transformers/utils/variant.py. The `variant` transformer still resolves from pts.transformers.variant. * pts: variant score normalisation rescales with the shared min_max_scaler normalise_score drops its own _rescale, which reproduced Spark's mixed single and double precision arithmetic, and rescales with min_max_scaler in double precision. AlphaMissense, CADD and GERP normalisedScore values can move in the last digits compared with the Spark output. * pts, orchestration: FoldX annotations run inside the variant step The FoldX transform is the variant step's first task, required by the variant index, in both the platform config and the PPP override, so the two no longer start a VM each. pts_foldx leaves the DAG and pts_variant depends on pis_variant directly; FoldX output is read only by the variant index. * pts: one variant identifier hashing rule for evidence, LoF curation and the variant index hash_variant_index_ids and hash_long_variant_ids disagreed on IDs of exactly 300 characters: evidence and LoF curation hashed only longer IDs, the variant index hashed at the threshold, so an evidence variantId of exactly 300 characters never joined the index. The two merge into hash_long_variant_ids with the variant index's rule. Coordinates come from columns when the caller has them, as LoF curation and the variant index do, and are otherwise extracted from the ID as evidence did. The only change to evidence output is that 300-character IDs are now hashed. * pts: GPN-Star scores in the variant index (#233) * pis: mirror GPN-Star (M) LLR scores from a pinned Hugging Face commit * pts: GPN-Star cut-offs, normalisation and assessment for the variant index * pts: GPN-Star cut-offs require a strictly ordered minimum * pts: read GPN-Star scores for index positions and add them to variant effects * pts: ignore unscored GPN-Star rows when measuring cut-offs * pts: the variant step adds GPN-Star entries to scored SNVs * pts, orchestration: configure GPN-Star for the variant step * pts: rechunk the GPN-Star join before appending the effect, so real-sized batches stop panicking A left join over more than ~500 rows returns a struct column split into one chunk per polars thread. In polars 2.0.0, when/then over concat_list of a multi-chunk struct panics (struct_/mod.rs:475, left == right). Rechunking the joined batch gives the expression a single chunk whatever the join returns. * pts, orchestration: pin GPN-Star cut-offs to the revision, warn on missing chromosomes, bin scores consistently * pts: drop the CADD variant effect, which no VEP run provides VEP runs without the CADD plugin, so cadd_phred is always null and the CADD normalisation never applies. SpliceAI and Pangolin were already gone from the polars index. The output schema is unchanged. Closes opentargets/issues#4591
orchestration to 26.12.0rc23, pis to 26.12.0rc10, pts to 26.12.0rc25
#234) gs://bigbrain_data/prod/20261029T162943/ is the run 20261029T125742 with pubmedId set to 41040686 and initialSampleSize to "10,725 samples from 4,656 donors (all ancestries)" in both study indexes, and the same credible sets. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
…ted prod run folder (#232) * orchestration, pis, pts: move the eqtl catalogue r8 release into a dated prod run folder The r8 run of 2026-08-22 now lives in gs://eqtl_catalogue_data/r8/prod/20260822T223458/ with its outputs under output/, its DAG config under config/ and its docs under docs/. - eqtl_catalogue_ingestion.yaml: outputs point at the run's output/ folder - pis/config.yaml: molqtl reads r8 from the run folder instead of r7, and no longer copies MAGE_OTAR, IBDverse and GTEx_v10_majiq on their own, as r8 has GTEx_v10, IBDverse and MAGE; molqtl_ppp no longer copies r8, as molqtl runs in PPP too - pts/config.yaml: study and credible set validation read r8 - docs: rewrite the eQTL Catalogue README for the run, with the inputs, gentropy links pinned to v3.4.0-dev.4, counts and output schemas - Makefile: upload-eqtl-catalogue-readme uploads the docs to a run folder given by BUCKET Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * pis, docs: read the eqtl catalogue r8 study index with pubmedId patched gentropy v3.4.0-dev.4 wrote pubmedId as a float string ("3.2913098E7") and "NaN" for AFR_LCL. gs://eqtl_catalogue_data/r8/prod/20261029T162303/ holds the run 20260822T223458 with pubmedId patched to whole numbers and nulls, and the same credible sets. PIS reads it; the README describes the patch. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * orchestration: use the renamed eqtl catalogue readme target in the Makefile upload-eqtl-catalogue-bucket-readme was renamed to upload-eqtl-catalogue-readme, but .PHONY and update-bucket-docs kept the old name, so update-bucket-docs skipped it silently. The new target needs BUCKET, so it is left out of update-bucket-docs, like the other run folder targets. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
orchestration to 26.12.0rc25, pis to 26.12.0rc11, pts to 26.12.0rc26
Merged
4 tasks done
project-defiant
marked this pull request as ready for review
October 31, 2026 20:19
…st steps end A cluster of a staging DAG was created after the uploads of the run, before its first steps. A cluster whose first step depends on a step of another cluster was therefore created when the run started, and stood idle until that step ended; past `idle_delete_ttl`, Dataproc deleted it before its first step. The create task now also follows the steps that the first steps of its cluster depend on. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
project-defiant
force-pushed
the
feat/staging-multiple-clusters
branch
from
November 2, 2026 09:42
9478699 to
995d5fc
Compare
Contributor
Author
There was a problem hiding this comment.
🟡 Changes recommended
An unresolved finding remains, flagging that the new and renamed upload-*-readme Makefile targets are missing from .PHONY unlike their siblings.
1 open issue
What changed in this PR
Despite its title (which describes the staging-DAG cluster-ordering change), this is a large bundled PR. Its substantive themes are: (1) a new staging-DAG framework where a Dataproc cluster's create task now waits on the upstream steps that its first steps depend on, so a cluster is created only when those steps end (avoiding idle-TTL deletion of a cluster created too early); (2) a Spark→Polars port of the PTS variant index with a new GPN-Star constraint-score annotation; and (3) migration of several input data sources to dated prod/<timestamp>/output/ run folders, with matching docs.
Changes:
- Staging DAGs:
build_staging_dagaddstasks[upstream] >> createfor each first-step dependency; newmodels/staging_dag/*schema and three new staging DAG packages (top hits, finngen multiome, bigbrain). - PTS variant index rewritten in Polars (
transformers/variant/*), PySpark variant modules deleted, GPN-Star scoring added (gpn_star.py, cut-offs in config), andhash_long_variant_idsunified (threshold now>=, explicit chromosome/position). - PIS/orchestration config: eQTL Catalogue r8 / FinnGen / GWAS Catalog sources repointed to
prod/<timestamp>/output/, Batch spec models hardened withextra='forbid', Makefile README-upload targets made run-folder aware.
| File | Description |
|---|---|
| orchestration/src/orchestration/utils/staging.py | Core change: cluster create follows first-step upstreams; delete after last step |
| orchestration/src/orchestration/models/staging_dag/*.py | New config schema (environment/infrastructure/dag/setup/placeholders) |
| orchestration/src/orchestration/dags/{gwas_catalog_top_hits,finngen_multiome_ingestion,bigbrain_ingestion}/* | New staging DAG packages + configs |
| orchestration/src/orchestration/dags/config/{unified_pipeline.yaml,clusters.yaml,*.yaml} | Remove pts_foldx/pts_variant cluster; repoint sources to run folders |
| pts/src/pts/transformers/variant/{init,index,gpn_star,consequences}.py | Polars variant index + GPN-Star annotation |
| pts/src/pts/transformers/utils/{variant,dataset}.py | Unified hash_long_variant_ids; shared list_parts |
| pts/src/pts/schemas/vep.py | Reduced VEP schema to the fields the index reads |
| pts/src/pts/pyspark/variant*.py | Deleted obsolete PySpark variant code |
| pts/config.yaml, orchestration/.../ppp/pts.override.yaml | Merge FoldX+variant step, add GPN-Star source/cut-offs |
| pis/config.yaml, pis/tests/test_variant_config.py | GPN-Star HF download; eQTL r8 / FinnGen source paths |
| pts/tests/*, orchestration/tests/* | New/updated tests for variant index, staging, configs |
| orchestration/Makefile, orchestration/docs/** | Run-folder README uploads; new/updated datasource docs & SVGs |
Notes: I verified the staging cluster logic against test_order_of_two_clusters, confirmed no lingering references to the deleted PySpark modules or plh, checked that the reduced vep_schema still covers every field read in vep.py, and that normalise_score handles every method produced (LOFTEE/GERP/AlphaMissense/VEP). The GPN-Star normalisation/cut-off math matches its tests. One minor .PHONY consistency issue is flagged inline.
Review effort: Balanced
| upload-finngen-bucket-readme upload-gwas-catalog-buckets-readme \ | ||
| upload-eqtl-catalogue-readme upload-ukb-ppp-bucket-readme \ | ||
| upload-finngen-gwas-readme upload-finngen-multiome-readme \ | ||
| upload-gwas-catalog-buckets-readme \ |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Based on
main. Replaces #236, closed by the rename of its branch.Summary
A cluster of a staging DAG was created after the uploads of the run, before its first steps. A cluster whose first step depends on a step of another cluster was therefore created when the run started, and stood idle until that step ended; past
idle_delete_ttl, Dataproc deleted it before its first step.The create task of a cluster now also follows the steps that the first steps of the cluster depend on, so the cluster is created when those steps end. A DAG with one cluster, such as the GWAS Catalog top hits, is unchanged.
This is the order of the GWAS Catalog summary statistics PICS DAG, whose window clumping and LD annotation run on two clusters: the cluster of
ld_annotationandpicsis created whenwindow_clumpingandstudy_indexend.Test plan
tests/test_staging_dag.py: a DAG with a step on a second cluster, created after the step on the first cluster and deleted after its own last steptests/test_staging_models.py,tests/test_dag_validation.py: 74 passed in total🤖 Generated with Claude Code