.. DO NOT EDIT.
.. THIS FILE WAS AUTOMATICALLY GENERATED BY SPHINX-GALLERY.
.. TO MAKE CHANGES, EDIT THE SOURCE PYTHON FILE:
.. "starterkit/Ex04_basics_b2luigi_averagetask.py"
.. LINE NUMBERS ARE GIVEN BELOW.

.. only:: html

    .. note::
        :class: sphx-glr-download-link-note

        :ref:`Go to the end <sphx_glr_download_starterkit_Ex04_basics_b2luigi_averagetask.py>`
        to download the full example code.

.. rst-class:: sphx-glr-example-title

.. _sphx_glr_starterkit_Ex04_basics_b2luigi_averagetask.py:


.. _exercise04_label:

Multiple Tasks and Introduction to Schedulers
=============================================

.. hint::
    This example demonstrates how a single task can require multiple tasks to complete.
    This is useful if the output of multiple tasks is needed to compute the output of the following task.
    Additionally, we introduce the core concepts of schedulers.

:class:`b2luigi.WrapperTask` is not the only way to require multiple tasks.
Also normal tasks can require multiple tasks.
This is useful if the wrapper task uses the inputs of the required tasks to compute its output.

For this, we define the first and second task in the same way as in the previous example.

.. GENERATED FROM PYTHON SOURCE LINES 18-49

.. code-block:: Python

    import b2luigi


    class MyTask(b2luigi.Task):
        parameter = b2luigi.IntParameter()

        def run(self):
            with open(self.get_output_file_name("output.txt"), "w") as f:
                f.write(f"{self.parameter}")

        def output(self):
            yield self.add_to_output("output.txt")


    class MyOtherTask(b2luigi.Task):
        parameter = b2luigi.IntParameter()

        def requires(self):
            return MyTask(parameter=self.parameter)

        def output(self):
            yield self.add_to_output("output2.txt")

        def run(self):
            with open(self.get_input_file_names("output.txt")[0], "r") as f:
                number = int(f.read())

            with open(self.get_output_file_name("output2.txt"), "w") as f:
                f.write(f"{number**2}")



.. GENERATED FROM PYTHON SOURCE LINES 50-55

In this example we will build a task whose ``require`` method and the task parameter look identical to the previous example in :ref:`exercise03_label`.

Now, we make use of the fact that the iterable of :meth:`b2luigi.Task.get_input_file_names`
returns all the input targets that match the provided key.
In this case, the key is the file name defined with :meth:`b2luigi.Task.add_to_output` in ``MyOtherTask``.

.. GENERATED FROM PYTHON SOURCE LINES 58-83

.. code-block:: Python

    class MyAverageTask(b2luigi.Task):
        max_value = b2luigi.IntParameter()

        def requires(self):
            return [MyOtherTask(parameter=value) for value in range(self.max_value)]

        def output(self):
            # Define multiple outputs at the same time.
            yield self.add_to_output("average.txt")
            yield self.add_to_output("sum.txt")

        def run(self):
            numbers = []

            for filename in self.get_input_file_names("output2.txt"):
                with open(filename, "r") as f:
                    numbers.append(int(f.read()))

            with open(self.get_output_file_name("average.txt"), "w") as f:
                f.write(f"{sum(numbers)/len(numbers)}")

            with open(self.get_output_file_name("sum.txt"), "w") as f:
                f.write(f"{sum(numbers)}")



.. GENERATED FROM PYTHON SOURCE LINES 84-87

The output of ``MyAverageTask`` will now be in the `result_dir`, however,
since we used the :meth:`b2luigi.Task.add_to_output` it will create another subdirectory
corresponding to its parameter value, i.e. `max_value=10`.

.. GENERATED FROM PYTHON SOURCE LINES 89-93

.. code-block:: Python

    if __name__ == "__main__":
        b2luigi.set_setting("result_dir", "results")
        b2luigi.process(MyAverageTask(max_value=10))


.. GENERATED FROM PYTHON SOURCE LINES 94-121

In real applications, tasks depending on multiple other tasks can
form very large dependency graphs. Additionally, the execution time of the
entire workflow can be very long. To handle the schedule and execution of
such workflows, ``luigi`` provides the user with the concept of local and
global schedulers. The ``local_scheduler`` is the default scheduler and is
run in the background without the user noticing it.

We can also define which scheduler should be used to schedule the task
here. For this, we spawn the scheduler in a separate process with the
help of the ``luigi`` daemon:

.. code-block:: bash

  host@user:$ luigid --port 8080

The ``port`` number should be an accessible port of the ``host`` machine. We
can then use the ``--scheduler-host`` and ``--scheduler-port`` arguments to
tell the task where to find the scheduler.

.. code-block:: bash

  otherhost@user:$ python3 Ex04_basics_b2luigi_averagetask.py --scheduler-host host.url --scheduler-port 8080

The luigi scheduler now can also be see when spectating the port
on the ``host`` machine in the browser with ``localhost:8080``.
The scheduler settings can also be set in the `settings.json` or
in the `luigi.cfg` file.


.. _sphx_glr_download_starterkit_Ex04_basics_b2luigi_averagetask.py:

.. only:: html

  .. container:: sphx-glr-footer sphx-glr-footer-example

    .. container:: sphx-glr-download sphx-glr-download-python

      :download:`Download Python source code: Ex04_basics_b2luigi_averagetask.py <Ex04_basics_b2luigi_averagetask.py>`

    .. container:: sphx-glr-download sphx-glr-download-zip

      :download:`Download zipped: Ex04_basics_b2luigi_averagetask.zip <Ex04_basics_b2luigi_averagetask.zip>`


.. only:: html

 .. rst-class:: sphx-glr-signature

    `Gallery generated by Sphinx-Gallery <https://sphinx-gallery.github.io>`_