Skip to content

feat: add support for REANA #231

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Alputer wants to merge 1 commit into
base: main
Choose a base branch
from
+38 −1
Open

feat: add support for REANA #231

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 30 additions & 0 deletions analyses/cms-open-data-ttbar/reana.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
inputs:
files:

@alexander-held alexander-held Feb 28, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A few of the files here aren't technically needed I think, like nanoad_branch_ratios.json, GetIOBranches.py, jetassignment_training* and make_corrections_json.

@Alputer Alputer Feb 28, 2025

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, indeed most of the inputs were unnecessary. Updated the inputs list.

- ttbar_analysis_pipeline.py
- nanoaod_inputs.json
- cabinetry_config.yml
- cabinetry_config_ml.yml
- corrections.json
directories:
- histograms
- utils
- models
- reference
workflow:
type: serial
resources:
dask:
image: registry.cern.ch/docker.io/alputer/agc-dask:1.0.0

@tiborsimko tiborsimko Feb 28, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the analysis has N_FILES_MAX_PER_SAMPLE = 5 by default, we don't have to have so much Dask workers.

Let's tune the parameters such as number_of_workers down so that the default reana.yaml would be executable even on a laptop.

(And we could recommend "big" settings for the full data processing somewhere in the docs.)

@Alputer Alputer Feb 28, 2025

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I forgot that N_FILES_MAX_PER_SAMPLE = 5 by default. I will scale down the cluster and update reana.yaml accordingly.

@Alputer Alputer Feb 28, 2025

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I scaled down the requested cluster size, but I guess with default settings it is not runnable locally even with N_FILES_MAX_PER_SAMPLE = 5. I tried the following combinations and the cluster was under a lot of pressure and execution was extremely slow. So I think we should use more powerful clusters in any case ...

Option 1 (Was very slow, at least half an hour execution time even with INPUT_FROM_EOS=True)

number_of_workers = 2
single_worker_memory = 2Gi

Option 2 (Workers under a lot of pressure, losing progress and likely to fail)

number_of_workers = 5
single_worker_memory = 1Gi

@tiborsimko tiborsimko Mar 5, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

FWIW here is one data point. Testing on a laptop with 12 CPU cores and 32 GB RAM, using INPUT_FROM_EOS being True. (So the run includes inference etc.) Here are results for 5 Dask workers of 2 GiB memory and 1 thread (execution time 15 min 40 sec) and then for 10 Dask workers (10 min 28 sec). (Kerberos was off in both cases; enabling it could give further speed-up.)

image

P.S. A further test with 15 workers showed similar time as with 10 workers, which makes perfect sense, because the laptop has only 12 cores.

number_of_workers: 20
single_worker_memory: 2Gi
single_worker_threads: 1
specification:
steps:
- name: agc
environment: registry.cern.ch/docker.io/alputer/agc-dask:1.0.0
commands:
- python3 ttbar_analysis_pipeline.py
outputs:
files:
- histograms.root

@alexander-held alexander-held Feb 28, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think workspace.json would be another good addition here as a product of the workflow (it contains the statistical model).

- workspace.json
7 changes: 7 additions & 0 deletions analyses/cms-open-data-ttbar/utils/clients.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,13 @@ def get_client(af="coffea_casa"):
cluster.scale(10)
client = cluster.get_client()

elif af == "reana":

@tiborsimko tiborsimko Feb 28, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is the new analysis facility necessary? Perhaps not if we do something like:

DASK_SCHEDULER_URI = os.getenv("DASK_SCHEDULER_URI", "tcp://127.0.0.1:8080")
client = dask.distributed.Client(DASK_SCHEDULER_URI)

I.e. read the environment variable to know where to connect, and if there is none set, then use the Dask defaults?

Such a technique could perhaps work for both REANA use case and the local use case. (Untested.)

One advantage this would bring is to "teach" researchers how to write Dask code runnable both locally and on REANA from the start.

@alexander-held alexander-held Feb 28, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it might be difficult to fully generalize the logic: on some facilities we expect to connect to an existing cluster, which may or may not have an addressed stored in an environment variable, but then if we do not find that the right action to take still depends on the facility as the spawning varies as well. I'm open to suggestions though.

We could add the logic at the beginning of the function to connect to an existing cluster if the environment variable points to one and ignore the value of af in this case but I'm not sure if that would be easier to follow.

@Alputer Alputer Feb 28, 2025

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think without introducing a new analysis facility, code will be less intuitive as all different ways to run and connect to Dask clusters are handled in seperate if clauses, and handling REANA and local use cases together will be an exception to that.

@tiborsimko tiborsimko Feb 28, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, I thought above only about "local" and "reana", that these two could perhaps be one. (Because the only thing that is different is the URI of the Dask scheduler.) But for the other analysis facilities, and their special spawning, would have to stay apart...

@tiborsimko tiborsimko Mar 5, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@alexander-held BTW it would be very nice to make the analysis fully configurable so that people wouldn't have to edit utils/config.py for things such as selecting analysis facility or whether inference should be done or not.

This could be done by two ways, either respecting command-line arguments to the notebook in the papermill style, or reading values from environment variables if they are set. For example:

# a) using command-line options
$ python3 ttbar_analysis_pipeline.py --use-inference False
# b) using environment variables
$ USE_INFERENCE=False python3 ttbar_analysis_pipeline.py

Would you be open for such changes to the repository? We could discuss IRL on Friday.

@alexander-held alexander-held Mar 5, 2025 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For the af handling: I would propose to stick with the implementation we have instead of partially merging logic as I'm afraid that may only make it more complicated to follow.

For the topic of configuring things: yes, that sounds like a good idea to me. There are a few things in the main file

analysis-grand-challenge/analyses/cms-open-data-ttbar/ttbar_analysis_pipeline.py

Lines 88 to 106 in c37bf59

### GLOBAL CONFIGURATION
# input files per process, set to e.g. 10 (smaller number = faster)
N_FILES_MAX_PER_SAMPLE = 5
# enable Dask
USE_DASK = True
# enable ServiceX, specify options
USE_SERVICEX = False
USE_SERVICEX_UPROOT_RAW = True # set False to use func_adl instead
USE_SERVICEX_DOWNLOAD = False # set False to use remote data access
### ML-INFERENCE SETTINGS
# enable ML inference
USE_INFERENCE = True
# enable inference using NVIDIA Triton server
USE_TRITON = False
as well as the options in utils/config.py. We probably do not need to support configuring all of them externally but the performance-related ones would be useful. A CLI feels more pythonic to me than picking up environment variables and would still allow env-based configuration by just feeding those variables in through the CLI. We might need to think a bit about how to integrate this in a way that is not too invasive to keep the notebook / .py synchronization working but I'm sure that we can find a solution there.

import os
from dask.distributed import Client

DASK_SCHEDULER_URI = os.getenv("DASK_SCHEDULER_URI")
client = Client(DASK_SCHEDULER_URI)

elif af == "local":
from dask.distributed import Client

Expand Down
2 changes: 1 addition & 1 deletion analyses/cms-open-data-ttbar/utils/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
# ServiceX: set False to use remote data access
"USE_SERVICEX_DOWNLOAD": False,
# analysis facility: set to "coffea_casa" for coffea-casa environments,
# "EAF" for FNAL, "purdue-af" for Purdue Analysis Facility, "local" for local setups
# "EAF" for FNAL, "purdue-af" for Purdue Analysis Facility, "reana" for REANA, "local" for local setups
"AF": "coffea_casa",
# number of bins for standard histograms in processor
"NUM_BINS": 25,
Expand Down