Hydra integration

This example shows how to build Pipeline object with Hydra using building blocks from spdl.pipeline.defs module.

The definition of the pipeline is found in "hydra_integration.yaml" file.

# Note: `_convert_: "all"` is required when using a factory function
# such as Pipe, Aggregate and Disaggregate.

_convert_: all

pipeline_cfg:
  _convert_: all
  _target_: spdl.pipeline.defs.PipelineConfig
  src:
    _target_: spdl.pipeline.defs.SourceConfig
    source:
      # range(20)
      _target_: builtins.range
      _args_: [ 20 ]
  pipes:
    # Aggregate(3, drop_last=True)
    - _target_: spdl.pipeline.defs.Aggregate
      _args_: [ 3 ]
      drop_last: true
    # Pipe(sum)
    - _target_: spdl.pipeline.defs.Pipe
      _args_:
        - _target_: builtins.sum
          _partial_: true
  sink:
    _target_: spdl.pipeline.defs.SinkConfig
    buffer_size: 3

pipeline:
  _convert_: all
  _target_: spdl.pipeline.build_pipeline
  _args_: [ "${pipeline_cfg}" ]
  num_threads: 1

Source

Source

Click here to see the source.
 1# Copyright (c) Meta Platforms, Inc. and affiliates.
 2# All rights reserved.
 3#
 4# This source code is licensed under the BSD-style license found in the
 5# LICENSE file in the root directory of this source tree.
 6
 7
 8"""This example shows how to build :py:class:`~spdl.pipeline.Pipeline`
 9object with Hydra using building blocks from :py:mod:`spdl.pipeline.defs`
10module.
11
12The definition of the pipeline is found in ``"hydra_integration.yaml"`` file.
13
14.. literalinclude:: ../../../examples/hydra_integration.yaml
15    :language: yaml
16
17"""
18
19__all__ = ["main"]
20
21import os
22
23import hydra
24from omegaconf import DictConfig
25
26os.environ["HYDRA_FULL_ERROR"] = "1"
27
28
29@hydra.main(version_base=None, config_path=".", config_name="hydra_integration")
30def main(cfg: DictConfig) -> None:
31    """The main entry point.
32
33    Args:
34        cfg: The configuration created from the ``"hydra_integration.yaml"`` file.
35    """
36    pipeline_cfg = hydra.utils.instantiate(cfg.pipeline_cfg)
37    print(pipeline_cfg)
38
39    pipeline = hydra.utils.instantiate(cfg.pipeline)
40    print(pipeline)
41
42    with pipeline.auto_stop():
43        for i, item in enumerate(pipeline.get_iterator(timeout=3)):
44            print(i, f"{item=}")
45
46
47if __name__ == "__main__":
48    main()

API Reference

Functions

main(cfg: DictConfig) → None[source]

The main entry point.

Parameters:

cfg – The configuration created from the "hydra_integration.yaml" file.