Create your first ETL in Luigi
Tags: posts, dataengineering, datapipeline, etl
Published: 2019-11-24 06:19:05

---
title: Create your first ETL in Luigi
published: true
date: 2019-11-18 12:31:22 UTC
tags: Posts,data engineering,data pipeline,etl
canonical_url: http://blog.adnansiddiqi.me/create-your-first-etl-in-luigi/
---

![](https://luigi.readthedocs.io/en/stable/_static/luigi.png)

_This post is the part of [Data Engineering Series](http://blog.adnansiddiqi.me/tag/data-engineering/)._

In previous posts, I discussed writing ETLs in Bonobo, Spark, and Airflow. In this post, I am introducing another ETL tool which was developed by _Spotify_, called **Luigi**.

Earlier I had discussed [here](http://blog.adnansiddiqi.me/develop-your-first-etl-job-in-python-using-bonobo/), [here](http://blog.adnansiddiqi.me/getting-started-with-apache-airflow/) and [here](http://blog.adnansiddiqi.me/create-your-first-etl-pipeline-in-spark-and-python/) about writing basic ETL pipelines. Bonobo is cool for write ETL pipelines but the world is not all about writing ETL pipelines to automate things. There are other use cases in which you have to perform tasks in a certain order once or periodically. For instance:

- Monitoring Cron jobs
- transferring data from one place to another.
- Automating your DevOps operations.
- Periodically fetching data from websites and update the database for your awesome price comparison system.
- Data processing for recommendation-based systems.
- Machine Learning Pipelines.

The possibilities are endless.

Before we move on further to implement _Luigi_ in our systems, let’s discuss what actually is Airflow and it’s terminologies.

## What is Luigi?

From the [Github page](https://github.com/spotify/luigi):

> Luigi is a Python (2.7, 3.6, 3.7 tested) package that helps you build complex pipelines of batch jobs. It handles dependency resolution, workflow management, visualization, handling failures, command line integration, and much more.

Let’s learn and understand the basic components and terminologies.

- **Target:-** In simple words, a target holds the output of a task. A target could be a local(e.g: a file), HDFS or RDBMS(MySQL etc)
- **Task:-** Task is something where the actual work takes place. A task could be independent or dependent. The example of a dependant task is dumping the data into a file or database. Before loading the data the data must be there by any mean(scraping, API, etc). Each task is represented as a Python Class which contains certain mandatory member functions. A task function contains the following methods:
  - **requires():- **This member function of the task class contains all the task instances that must be executed before the current task. In the example I shared above, a task, named **ScrapeData** , will be included in the `requires()` method, hence make a task a dependant task.
  - **output():-** This method contains the target where the task output will be stored. This could contain one or more target objects.
  - **run():-** This method contains the actual logic to run a task.

The pictorial representation will be something like below:

[![](http://blog.adnansiddiqi.me/wp-content/uploads/2019/11/Luigi_FlowDiagram-300x128.jpg)](http://blog.adnansiddiqi.me/wp-content/uploads/2019/11/Luigi_FlowDiagram.jpg)

 

Let’s write a toy ETL first. It is doing nothing but put **Hello World** in a text file and then replace World with your input name.

```
import timeimport luigi# Task A - write hello world in text fileclass HelloWorld(luigi.Task): def requires(self): return None def output(self): return luigi.LocalTarget('helloworld.txt') def run(self): time.sleep(15) with self.output().open('w') as outfile: outfile.write('Hello World!\n') time.sleep(15)# Task B - pick the text from helloworld.txt, replace World with the input nameclass NameSubstituter(luigi.Task): name = luigi.Parameter() def requires(self): return HelloWorld() def output(self): return luigi.LocalTarget(self.input().path + '.name\_' + self.name) def run(self): time.sleep(15) with self.input().open() as infile, self.output().open('w') as outfile: text = infile.read() text = text.replace('World', self.name) outfile.write(text) time.sleep(15)if \_\_name\_\_ == '\_\_main\_\_': luigi.run()
```

The very first Class or the Task,`HelloWorld`, is the _extract_ part of the ETL, assume the text Hello World! is coming from an external source(API, DB, etc) and being stored in the file `helloworld.txt`. The `output()` method sets the target. Since the target is a local file `LocalTarget` with the file name `helloworld.txt`. The `run` method is taking care of all the processing logic thing. Since this task is not depending on any task hence `requires()` returns a `None`.

The second class `NameSubstituter` can be assumed a class transforming the original text into something else and saving into another text file. So, this class(task) is taking care of both **T** and **L** part of the ETL.

`name = luigi.Parameter()` is to parameterized the ETL which facilitates to accept data from the external data resources. The `infile.read()` reads the data from the incoming file, in our case, it is `helloworld.txt` and the content _Hello World!_ is being saved in a `text` variable. The text _World_, then being replaced by the input name. Also the file name is following a certain format which is being set in the `output()` method.

Alright, the code is ready. It’s time to run it. I go to the command line and run the following command:

`python luigitutorial.py --scheduler-host localhost NameSubstituter`

Oops! it crashed!

```
LearningLuigi python luigitutorial.py --scheduler-host localhost NameSubstituter Traceback (most recent call last): File "luigitutorial.py", line 48, in <module> luigi.run() File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/luigi/interface.py", line 194, in run luigi\_run\_result = \_run(\*args, \*\*kwargs) File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/luigi/interface.py", line 211, in \_run return \_schedule\_and\_run([cp.get\_task\_obj()], worker\_scheduler\_factory) File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/luigi/cmdline\_parser.py", line 114, in get\_task\_obj return self.\_get\_task\_cls()(\*\*self.\_get\_task\_kwargs()) File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/luigi/task\_register.py", line 88, in \_\_call\_\_ param\_values = cls.get\_param\_values(params, args, kwargs) File "/Users/AdnanAhmad/Data/anaconda3/lib/python3.7/site-packages/luigi/task.py", line 420, in get\_param\_values raise parameter.MissingParameterException("%s: requires the '%s' parameter to be set" % (exc\_desc, param\_name))luigi.parameter.MissingParameterException: NameSubstituter[args=(), kwargs={}]: requires the 'name' parameter to be set
```

As you can see, the error message is clear: _requires the ‘name’ parameter to be set._

We’d have to pass the name parameter. For that we will do the following:

`python luigitutorial.py --scheduler-host localhost NameSubstituter --name Adnan`

![](http://blog.adnansiddiqi.me/wp-content/uploads/2019/11/Screenshot-2019-11-18-at-4.24.29-PM.png)

Can you see the smiley sign? It all goes well!

Now let me explain the command. Luigi uses a different kind of schedulers to schedule the jobs. For development purpose `--local-schedular` is used but if you want to visualize the process for the monitoring purpose than you should use `--schedular-host` to monitor it on a web-based interface. Make sure you run the `luigid` daemon which runs a local webserver. If you do not set `--schedular-host` it will still run but you can’t monitor the running task so DO take care of that! If all dots are connected well, you can see how the things going by visiting `http://localhost:8082/`

On a visit you can see screens like below:

![](http://blog.adnansiddiqi.me/wp-content/uploads/2019/11/Screenshot-2019-11-18-at-4.16.14-PM.png)

![](http://blog.adnansiddiqi.me/wp-content/uploads/2019/11/Screenshot-2019-11-18-at-4.37.49-PM.png)

The tasks are running and you can see the status. In case you wonder why I added `sleep`, you can guess now. Without adding a delay you could not visualize it as it would execute very fast. Also notice multiple entries of `NameSubstituter` class with parameter in _Details_ section. It is because they were considered unique jobs while _HelloWorld_ was not.

![](http://blog.adnansiddiqi.me/wp-content/uploads/2019/11/Screenshot-2019-11-18-at-4.17.27-PM.png)

If you click an individual task you can see the dependency graph of tasks. Similar to [Airflow](http://blog.adnansiddiqi.me/getting-started-with-apache-airflow/).

```
-rw-r--r-- 1 AdnanAhmad staff 13 Nov 18 16:16 helloworld.txt-rw-r--r-- 1 AdnanAhmad staff 13 Nov 18 16:22 helloworld.txt.name\_Adnan-rw-r--r-- 1 AdnanAhmad staff 13 Nov 18 16:16 helloworld.txt.name\_Jerry
```

You see file names appended by the input name. If you remember we had set the file name like that. It is not necessary, you can pick whatever you want as I myself took it from an example.

Oh, by the way, the tasks are run once. It is not like files are generated on every run. If you want to take a fresh start than just delete all input and output files. In this example, if you, for instance, want to re-run the ETL with the name Adnan then just simply remove `helloworld.txt.name_Adnan`, not all files. If your input file content is changed then remove that as well.

## Conclusion

So you learned how Luigi makes it easier to write ETLs for your needs. In the next part, we will be discussing a real-world example as we had done for Apache Airflow. Like always, the code is available on [Github](https://github.com/kadnan/Luigi-Python-Tutorial).