Luigi orchestrator

2.1. How it works

Luigi work with tasks that are defined with classes that extend the luigi.Task class. The basic usage is to overwrite the run function (with what to do) and the output (with the file that will output).

For example a basic task would look like:

import luigi
from datetime import date, datetime

class ReportTask(luigi.Task):

    def run(self):
        with open(f"output.txt", "w") as stream:
            stream.write(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

    def output(self):
        return luigi.LocalTarget("output.txt")

This task will write a file called output.txt with the execution datetime.

2.2. Basic example

For a real example tasks should also have dependencies.

First of all, for this example let’s create two python files (register.py and report.py) which run simple python task. They both will accept a name as parameter so that it can be run for different days. The first one will simply write a file with the execution time:

register.py

from datetime import date, datetime

def main(filename):

    uri = f"output/{filename}.txt"

    with open(uri, "w") as stream:
        stream.write(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

    print(f"File '{uri}' wrote")

And the other will write the same as an html file:

report.py

from datetime import date, datetime
from markdown import markdown


def main(filename):

    uri = f"output/{filename}.html"

    html = markdown(
        f"""# Report
        {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
        """
    )

    with open(uri, "w") as stream:
        stream.write(html)

    print(f"File '{uri}' wrote")

With those files it is possible to create the associated luigi tasks:

from datetime import date, datetime
import luigi

class RegisterTask(luigi.Task):

    mdate = luigi.DateParameter(default=date.today())

    def run(self):
        from register import main

        main(self.mdate.strftime("%Y_%m_%d"))

    def output(self):
        return luigi.LocalTarget(self.mdate.strftime("output/%Y_%m_%d.txt"))


class ReportTask(luigi.Task):

    mdate = luigi.DateParameter(default=date.today())

    def run(self):
        from report import main

        main(self.mdate.strftime("%Y_%m_%d"))

    def output(self):
        return luigi.LocalTarget(self.mdate.strftime("output/%Y_%m_%d.html"))

Both tasks have a parameter called mdate so that each day the task runs it will write a different file.

Finally let’s add a dummy task that requires both RegisterTask and ReportTask.

class DoAllTask(luigi.WrapperTask):

    mdate = luigi.DateParameter(default=date.today())

    def requires(self):
        return RegisterTask(self.mdate), ReportTask(self.mdate)

To run the example you first need to start the luigi server with:

luigid

And then you put the three tasks to a file called master.py and add:

if __name__ == "__main__":
    luigi.build([DoAllTask()])

You can run:

python master.py

And you should see how Luigi completes the 3 tasks.

2.3. Luigi server

If you go to localhost:8082 you can see luigi server.

It gives a summary of the tasks and their status.

If you select a task (for example DoAllTask) you can see the dependency tree.

It is also possible view the dependency graph created with D3.

4.1. Install supervisor

First install it with:

sudo apt install supervisor -y

Then create some auxiliar folders that you will need:

#!/bin/bash 

# For luigi logs
sudo mkdir /var/log/luigi/

# For cron logs (and give writte permission)
sudo mkdir /var/log/cron/
sudo chmod 777 /var/log/cron/

# This is to allow luigi to store the worker state
sudo mkdir /var/lib/luigi-server/

And now declare the luigid service by creating the file /etc/supervisor/conf.d/luigid.conf

/etc/supervisor/conf.d/luigid.conf

[program:luigid]
command=/home/ubuntu/.local/bin/luigid
stopsignal=QUIT
stdout_logfile=/var/log/luigi/luigid.log
stderr_logfile=/var/log/luigi/luigid.error.log
autorestart=true
user=ubuntu

It is important to use the full path for the command that supervisor will run. Be careful to set user to one that has visbility of the luigid file.

Then start supervisor:

sudo supervisorctl reread
sudo service supervisor restart

# Check the result
sudo supervisorctl status

If there are no errors you can start using Luigi.

4.2. Create a virtual environment

First install virtualenv with:

pip3 install virtualenv

Then create the virtual environment:

# Make a folder for virtual environments
mkdir /home/ubuntu/venv/

# Create a virtualenv called vtasks
virtualenv -p /home/ubuntu/venv/ vtasks

Then to activate the environment you can call:

source /home/ubuntu/venv/vtasks/bin/activate

And deactivate to deactivate the environment.

4.3. Script file to run luigi

Let’s create the run_luigi.sh file with:

~/run_luigi.sh

#!/bin/bash 

# Add ssh keys
eval `keychain --agents ssh --eval github_ssh`

# Go to desired path
cd /home/ubuntu/villoro_tasks/

# Activate virtual environment
source /home/ubuntu/venv/vtasks/bin/activate

# Git fetch and checkout
git fetch
git checkout master
git pull origin master

# Install requirements
pip install -r requirements.txt

# Run luigi
python src/master.py

deactivate

This script does some things:

1. Loading the ssh keys so that it can work with GIT.
2. Change the path to the repository
3. Activate the virtual environment
4. GIT fetch and checkout new changes in master
5. Install requirements
6. Run Luigi

In order to run this you should give it executable permissions with:

chmod +x /home/ubuntu/run_luigi.sh

4.4. Runing luigi tasks with cron

Te best way to run Luigi automatically is using cron.

After doing all the previous steps you can set cron to run the script with:

crontab -e

Add this adding this line at the end:

0 4 * * * /home/ubuntu/run_luigi.sh >> /var/log/cron/luigi.log 2>&1

In this example it will run each day at 04:00 am GMT. If you want to run at different moment you can use Crontab.guru to check cron expressions.

It will also write the output as log file called /var/log/cron/luigi.log.