Does new coffea with DaskExecutor function with BaseSchema?

[tomeichlersmith]

I ask this seemingly absurd question because I noticed (with the runner configuration below), that my processor does not even get to the process call before a dask error is thrown.

    runner = processor.Runner(
        executor = processor.DaskExecutor(
            client = client,
            compression = None,
            retries=0,
            worker_affinity = True
        ),
        pre_executor = processor.IterativeExecutor(),
        schema = BaseSchema,
        processor_compression = None,
        metadata_cache = {},
    )

Dask Error

I can copy in the full error, but the rest seems to be from the other workers being killed.

<dask.highlevelgraph.HighLevelGraph object at 0x7f18dec657e0>                                                         
 0. num-9649a8c34eb4f5de4ee14279d985e6d5                                                                              
>.                                                                                                                    
Traceback (most recent call last):                                                                                    
  File "/export/scratch/users/eichl008/ldmx/eat/v14/analysis/coffea/.local/lib/python3.10/site-packages/distributed/protocol/pickle.py", line 63, in dumps
    result = pickle.dumps(x, **dump_kwargs)
AttributeError: Can't pickle local object 'NanoEventsFactory.events.<locals>.<lambda>'

Am I mis-configuring the DaskExecutor? Do I need to use a different schema in order to avoid the local objects produced by the NanoEventsFactory?

[lgray]

Are you doing this in coffea 2023?

[lgray]

Ah, yes it is.

[tomeichlersmith]

Yes

coffea==2023.10.0rc1

pip freeze

awkward==2.5.0                                                                                                                                                                                                                                
awkward-cpp==26                                                                                                                                                                                                                               
awkward0==0.15.5  
boost-histogram==1.4.0
cachetools==5.3.2
click==8.1.7
cloudpickle==3.0.0
coffea==2023.10.0rc1
contourpy==1.2.0
correctionlib==2.4.0
cycler==0.12.1
dask==2023.11.0
dask-awkward==2023.11.3
dask-histogram==2023.10.0
distributed==2023.11.0
fonttools==4.44.3
fsspec==2023.10.0
hist==2.7.2
histoprint==2.4.0
importlib-metadata==6.8.0
Jinja2==3.1.2
kiwisolver==1.4.5
llvmlite==0.41.1
locket==1.0.0
lz4==4.3.2
markdown-it-py==3.0.0
MarkupSafe==2.1.3
matplotlib==3.8.1
mdurl==0.1.2
memory-profiler==0.61.0
mplhep==0.3.31
mplhep-data==0.0.3
msgpack==1.0.7
numba==0.58.1
numpy==1.23.5
packaging==23.2
pandas==2.1.3
partd==1.4.1
Pillow==10.1.0
psutil==5.9.6
pyarrow==14.0.1
pydantic==1.10.13
Pygments==2.16.1
pyparsing==3.1.1
python-dateutil==2.8.2
pytz==2023.3.post1
PyYAML==6.0.1
rich==13.7.0
scipy==1.11.3
six==1.16.0
sortedcontainers==2.4.0
tblib==3.0.0
toml==0.10.2
toolz==0.12.0
tornado==6.3.3
tqdm==4.66.1
typing_extensions==4.8.0
tzdata==2023.3
uhi==0.4.0
uproot==5.1.2
uproot3==3.14.4
uproot3-methods==0.10.1
urllib3==2.1.0
zict==3.0.0
zipp==3.17.0

[lgray]

The Runner interface is being made obsolete, largely by the adoption of dask-awkward.

What you should do instead is:

events = NanoEventsFactory.from_root(your_files, options, ....)

proc = YourProcessor(*args)
out_collections = proc(events)
result = dask.compute(out_collections)

We are making things over in #882 to deal better with the concept of datasets, you can merge that as well and use those interfaces if you like.

[lgray]

Detection of the Client is done automatically through dask.

If you want to use a specific client you should use a context manager:

with Client(<arguments>) as client:
    events = NanoEventsFactory.from_root(...)
    out, = dask.compute(analysis(events))

and that will ensure the client you specify in the with block is used for execution.

[lgray]

In coffea 2023 and onwards the definition of the analysis is separate from its execution, so there's so "connection" between nanoevents and creating the definition of your analysis and the Client at all.

[green-cabbage]

I see, does that mean that we define the chunk_size and maxchunks in metadata? Suppose that I want to set chunk_size = 100 and maxchunks = 2, I should have the fileset.json file as (for e.g):
"/DYJetsToLL_M-50_TuneCP5_13TeV-amcatnloFXFX-pythia8/RunIISummer20UL18NanoAODv9-106X_upgrade2018_realistic_v16_L1v1-v2/NANOAODSIM": { "files": { "root://grid-cms-xroo....20E.root": { "object_path": "Events", "steps": [ [ 0, 100 ], [ 100, 200 ], ], "uuid": <some id> }

Does this sound about right?

[lgray]

Use dataset_tools.preprocess and dataset_tools.max_chunks. That'll get you what you want.

The tests are actually pretty readable so just have a look there. Docs will come :-)

[lgray]

Actually, we now have some example based documentation for this. At least in terms of use in some simple but realistic examples. https://github.com/CoffeaTeam/coffea/blob/master/binder/processor.ipynb