Aggregation
This chapter provides information about passing aggregates of jobs to operation functions.
Definition
An aggregator
creates generators of aggregates for use in operation functions via FlowProject.operation
.
Such functions may accept a variable number of positional arguments, *jobs
.
The argument *jobs
is unpacked into an aggregate, defined as an ordered tuple of jobs.
See also the Python documentation about argument unpacking.
# project.py
from flow import FlowProject, aggregator
class Project(FlowProject):
pass
@Project.operation(aggregator=aggregator())
def op1(*jobs):
print("Number of jobs in aggregate:", len(jobs))
@Project.operation
def op2(job):
pass
if __name__ == "__main__":
Project().main()
If aggregator
is used with the default arguments, it will create a single aggregate containing all the jobs present in the project.
In the example above, op1
is an aggregate operation where all the jobs present in the project are passed as a variable number of positional arguments (via *jobs
), while op2
is an operation where only a single job is passed as an argument.
Tip
The concept of aggregation may be easier to understand if one realizes that “normal” operation functions are equivalent to aggregate operation functions with an aggregate group size of one job.
Types of Aggregation
Currently, signac-flow allows users to aggregate jobs in the following ways:
All jobs: All of the project’s jobs are passed to the operation function.
Group by state point key(s): The aggregates are grouped by one or more state point keys.
Group by arbitrary key function: The aggregates are grouped by keys determined by a key-function that expects an instance of
Job
and return the grouping key.Grouping into aggregates of a specific size.
Using a completely custom aggregator function when even greater flexibility is needed.
Group By
groupby()
allows users to aggregate jobs by grouping them by a state point key, an iterable of state point keys whose values define the groupings, or an arbitrary callable of Job
.
@Project.operation(aggregator=aggregator.groupby("temperature"))
def op3(*jobs):
pass
In the above example, the jobs will be aggregated based on the state point key "temperature"
.
So, all the jobs having the same value of temperature in their state point will be aggregated together.
Groups Of
groupsof()
allows users to aggregate jobs by generating aggregates of a given size.
@Project.operation(aggregator=aggregator.groupsof(2))
def op4(job1, job2=None):
pass
In the above example, the jobs will get aggregated in groups of 2 and hence, up to two jobs will be passed as arguments at once.
Note
In case the number of jobs in the project in this example is odd, there will be one aggregate containing only a single job.
In general, the last aggregate from groupsof()
will contain the remaining jobs if the aggregate size does not evenly divide the number of jobs in the project.
If a remainder is expected and valid, users should make sure that the operation function can be called with the reduced number of arguments (e.g. by using *jobs
or providing default arguments as shown above).
Sorting jobs for aggregation
Aggregators allow users to sort the jobs before creating aggregates with the sort_by
parameter.
The sorting order can be defined with the sort_ascending
parameter.
By default, when no sort_by
parameter is specified, the order of the jobs will be decided by the iteration order of the signac project.
@Project.operation(
aggregator=aggregator.groupsof(2, sort_by="temperature", sort_ascending=False)
)
def op5(*jobs):
pass
Note
In the above example, all the jobs will be sorted by the state point parameter "temperature"
in descending order and then be aggregated as groups of 2.
Selecting jobs for aggregation
signac-flow allows users to selectively choose which jobs to pass into operation functions. This can be used to generate aggregates from only the selected jobs, excluding any jobs that do not meet the selection criteria.
@Project.operation(aggregator=aggregator(select=lambda job: job.sp.temperature > 0))
def op6(*jobs):
pass
Aggregate ID
Similar to the concept of a job id, an aggregate id is a unique hash identifying an aggregate of jobs. The aggregate id is sensitive to the order of the jobs in the aggregate.
Note
The id of an aggregate containing one job is that job’s id.
In order to distinguish between an aggregate id and a job id, the id of aggregates with more than one job will always have a prefix agg-
.
Users can generate the aggregate id of an aggregate using flow.get_aggregate_id()
.
Tip
Users can also pass an aggregate id to the --job-id
command line flag provided by signac-flow in run
, submit
, and exec
.
Aggregation with FlowGroups
In order to associate an aggregator object with a FlowGroup
, signac-flow provides a group_aggregator
parameter in make_group()
.
By default, no aggregation takes place for a FlowGroup
.
Note
All the operations in a FlowGroup
will use the same aggregator
object provided to the group’s group_aggregator
parameter.
# project.py
from flow import FlowProject, aggregator
class Project(FlowProject):
pass
group = Project.make_group("agg-group", group_aggregator=aggregator())
@group
@Project.operation(aggregator=aggregator())
def op1(*jobs):
pass
@group
@Project.operation
def op2(*jobs):
pass
if __name__ == "__main__":
Project().main()
In the above example, when the group agg-group
is executed using python project.py run -o agg-group
, all the jobs in the project are passed as positional arguments for both op1
and op2
.
If op1
is executed using python project.py run -o op1
, all the jobs in the project are passed as positional arguments because a aggregator
is associated with the operation function op1
(separately from the aggregator used for agg-group
).
If op2
is executed using python project.py run -o op2
, only a single job is passed as an argument because no aggregator
is associated with the operation function op2
.