Merge pull request #480 from lgray/remove_lambdas

chore: various bits of housekeeping

src/dask_awkward/awkward.yaml

@@ -56,3 +56,8 @@ awkward:
     # continue on to be computed without dask-awkward specific
     # optimizations.
     on-fail: raise
+
+  aggregation:
+    # For tree reductions in dask-awkward, control how many partitions
+    # are aggregated per non-leaf tree node.
+    split-every: 8

src/dask_awkward/lib/core.py

@@ -2246,7 +2246,7 @@ def non_trivial_reduction(
     finalize_fn = partial(finalize_fn, **finalize_kwargs)
 
     if split_every is None:
-        split_every = 8
+        split_every = dask.config.get("awkward.aggregation.split-every", 8)
     elif split_every is False:
         split_every = sys.maxsize
     else:

src/dask_awkward/lib/io/io.py

@@ -8,11 +8,13 @@
 from typing import TYPE_CHECKING, Any, cast
 
 import awkward as ak
+import dask.config
 import numpy as np
 from awkward.types.numpytype import primitive_to_dtype
 from awkward.typetracer import length_zero_if_typetracer
 from dask.base import flatten, tokenize
 from dask.highlevelgraph import HighLevelGraph
+from dask.local import identity
 from dask.utils import funcname, is_integer, parse_bytes
 from fsspec.utils import infer_compression
 
@@ -651,7 +653,7 @@ def from_map(
                 axis=0,
             )
 
-            split_every = 8
+            split_every = dask.config.get("awkward.aggregation.split-every", 8)
 
             rep_trl_label = f"{label}-report"
             rep_trl_token = tokenize(result, second, concat_fn, split_every)
@@ -667,8 +669,8 @@ def from_map(
                 name_input=rep_part.name,
                 npartitions_input=rep_part.npartitions,
                 concat_func=concat_fn,
-                tree_node_func=lambda x: x,
-                finalize_func=lambda x: x,
+                tree_node_func=identity,
+                finalize_func=identity,
                 split_every=split_every,
                 tree_node_name=rep_trl_tree_node_name,
             )