ETL dengan luigi python

Luigi adalah package module dari python untuk preprocessing data yang dibagi secara perkelompok atau disebut batch processing.

Luigi pada awalnya dikembangkan oleh spotify untuk batch processing, yang membantu memberikan musik rekomendasi kepada pengguna, dan pengurutan daftar musik populer.

Untuk mempelajari luigi lebih lanjut, bisa membaca dokumentasi dari luigi disini.

Tutorial Luigi

Untuk pengaturan luigi, menggunakan ubuntu shell versi 20.04. Untuk upgrade ubuntu melalui command line dapat dilihat pada artikel berikut.

Selain itu, disini menggunakan python versi 3.7, disarankan versi python 3.6 keatas, dengan virtual environtment yang sudah terinstall. Untuk penginstallan virtual environtment bisa dilihat pada artikel berikut.

Penginstallan Luigi

Pertama siapkan project directory pada ubuntu melalui ubuntu shell dan lakukan penginstallan pada directory tersebut:

mkdir luigi-project
cd luigi-project

Buat virtual environtment untuk luigi dan aktifkan:

python3 -m venv luigi-venv
. luigi-venv/bin/activate

Virtual environtment terindikasi aktif apabila nama virtual environtment yang telah dibuat muncul pada tutorial ini nama dari virtual environtment adalah (luigi-venv).

Apabila virtual environtment telah aktif, terakhir adalah menginstall luigi untuk dapat digunakan pada proses ETL data. Penginstallan dapat menggunakan pip.

pip install luigi

Apabila penginstallan sukses, maka library luigi sudah bisa digunakan.

ETL menggunakan luigi

Pada tutorial ini akan menggunakan luigi untuk pemrosesan ETL menggunakan data pada project gutenberg , untuk mendapatkan list data e-book terpopuler. Secara konsep hampir mirip dengan proses ETL untuk mendapatkan data terpopuler yang digunakan spotify, untuk tutorialnya bisa dilihat dari dokumentasi luigi.

Menggunakan data project gutenberg, tujuan dari proses adalah untuk mendapatkan data berupa kata-kata yang paling umum digunakan pada buku-buku terpopuler.

Secara garis besar, berikut skema proses ETL yang akan dilakukan

Membuat task untuk ekstraksi

Pertama siapkan directory data sebagai tempat untuk menyimpan data yang diekstraksi. Pada directory data siapkan folder download untuk penyimpanan data yang akan di download dan counts untuk penyimpanan hasil data yang akan discraping.

mkdir data
mkdir data/counts
mkdir data/downloads

Berikutnya kita akan melakukan proses scraping dari website project gutenberg. Proses scraping memerlukan request library untuk mengunduh format html dan BeautifulSoup library untuk parsing data berformat html atau xml. Install library yang diperlukan untuk dapat digunakan pada proses ekstraksi data.

pip install wheel beautifulsoup4 requests

Berikutnya membuat luigi task file yang merupakan eksekusi dari pipeline data. Bisa dibuat secara manual dengan membuat file berformat python. Pada tutorial ini kita menamai file task dengan extract-link.py.

nano extract-link.py 

Pada file extract-link, tuliskan code sebagai berikut:

import requests
import luigi
from bs4 import BeautifulSoup
class GetTopBooks(luigi.Task):
 “””
 Get list of the most popular books from Project Gutenberg
 “””
def output(self):
 return luigi.LocalTarget(“data/books_list.txt”)
def run(self):
 resp = requests.get(“http://www.gutenberg.org/browse/scores/top")
soup = BeautifulSoup(resp.content, “html.parser”)
pageHeader = soup.find_all(“h2”, string=”Top 100 EBooks yesterday”)[0]
 listTop = pageHeader.find_next_sibling(“ol”)
with self.output().open(“w”) as f:
 for result in listTop.select(“li>a”):
 if “/ebooks/” in result[“href”]:
 f.write(“http://www.gutenberg.org{link}.txt.utf-8\n”
  .format(
 link=result[“href”]
 )
 )

Simpan file dan execute luigi task menggunakan command berikut:

python -m luigi — module extract-link GetTopBooks — local-scheduler

Luigi akan menghasilkan output execution summary berhasil.

Pada directory data luigi akan file ‘books_lists.txt’, sesuai method output yang dieksekusi pada luigi task. Apabila proses berhasil, maka kita sudah berhasil menggunakan luigi untuk mengekstrak data yang discraping dari website project gutenberg.

Menjalankan Luigi scheduler

Luigi scheduler berfungsi untuk memvisualisasikan proses eksekusi task. Untuk menjalankan interface luigi scheduler, aktifkan port 8082.

sudo ufw allow 8082/tcp

Jalankan scheduler dengan command sebagai berikut:

sudo sh -c “. luigi-venv/bin/activate ;luigid — background — port 8082”

Buka interface dari luigi scheduler menggunakan browser dengan mengakses server pada port 8082. Pada url browser : http://your_server_ip:8082, atau bisa menggunakan http://localhost:8082.

Luigi interface akan tampil sebagai berikut:

Eksekusi luigi task menggunakan central scheduler dengan command:

python -m luigi — module extract-link GetTopBooks

Pada luigi scheduler interface akan menampilkan proses dari task yang dieksekusi.

Task download data

Karena tujuan dari ETL untuk mendapatkan data kata-kata terpopuler pada daftar buku terpopuler. Maka akan ditambahkan luigi task untuk mendownload file buku dan proses pembersihan data dan penghitungan kata.

Pada file extract-link, tambahkan task downloadbooks() berdasarkan data dari task GetTopBooks yang telah didapatkan. Ingat saat menambahkan luigi task, tambahkan method yang akan diperlukan pada task.

. . .
class DownloadBooks(luigi.Task):
 “””
 Download a specified list of books
 “””
 FileID = luigi.IntParameter()
#untuk pemfilteran data
REPLACE_LIST = “””.,”’;_[]:*-”””
def requires(self):
 return GetTopBooks()
def output(self):
 return luigi.LocalTarget(“data/downloads/{}.txt”.format(self.FileID))
def run(self):
#membuka file data dari hasil task GetTopBooks()
 with self.input().open(“r”) as i:
 URL = i.read().splitlines()[self.FileID]
#mendownload dengan url yang ada pada list, menggunakan parameter FileID dan requests library
with self.output().open(“w”) as outfile:
 book_downloads = requests.get(URL)
 book_text = book_downloads.text
#Pemfilteran karakter selain huruf yang tidak diperlukan untuk load data kata terbanyak yang digunakan pada buku terpopuler
for char in self.REPLACE_LIST:
 book_text = book_text.replace(char, “ “)
#Transformasi kata menjadi lower case untuk penghitungan kata
book_text = book_text.lower()
 outfile.write(book_text)

Pada task DownloadBooks() adanya penggunaan parameter untuk pemrosesan download buku dari list topbook dengan id berbeda sesuai url yang telah didapatkan.

Eksekusi luigi task seperti sebelumnya.

python -m luigi — module word-frequency DownloadBooks — FileID 2

Apabila sukses, akan menampilkan output luigi execution summary succesfully.

Berikutnya adalah tahap akhir transformasi sebelum me load data yang sesuai dengan tujuan ETL.

Penghitungan kata

Pada data yang telah di download dan dibersihkan, akan dihitung kata terpopuler dari buku terpopuler.

Untuk pembacaan data, kita akan menggunakan library pickle dari python dan counter untuk penghitungan kata. Tambahkan pada file extract-link.py, dibawah baris import luigi, library tersebut.

from collections import Counter
import pickle

Kemudian tambahkan luigi task penghitungan kata.

class CountWords(luigi.Task):
 “””
 Count the frequency of the most common words from a file
 “””
FileID = luigi.IntParameter()
def requires(self):
 return DownloadBooks(FileID=self.FileID)
def output(self):
#membuka file dari hasil task sebelumnya
 return luigi.LocalTarget(
 “data/counts/count_{}.pickle”.format(self.FileID),
 format=luigi.format.Nop
 )
def run(self):
#penghitungan kata
 with self.input().open(“r”) as i:
 word_count = Counter(i.read().split())
#menggunakan pickle menyimpan data hasil penghitungan
with self.output().open(“w”) as outfile:
 pickle.dump(word_count, outfile)

Jalankan task dan monitor task pada center scheduler

python -m luigi — module word-frequency CountWords — FileID 2

Pada graph flow task akan terlihat seperti berikut:

Berikutnya kita akan melakukan penghitungan terhadap semua data yang telah di download, berarti akan ada proses task bertahap berdasarkan buku.

Untuk melakukannya membutuhkan parameter global dalam membaca dari data yang telah di ekstraksi untuk di transformasi sebelum penghitungan.

Tambahkan class untuk GlobalParam pada extract-link.py file

class GlobalParams(luigi.Config):
    NumberBooks = luigi.IntParameter(default=10)
    NumberTopWords = luigi.IntParameter(default=500)

Tambahkan pada file, luigi task TopWords untuk mengaggregasi data dari tiap file buku yang telah diekstraksi.

class TopWords(luigi.Task):
    """
    Aggregate the count results from the different files
    """
def requires(self):
#iterasi penghitungan kata berdasarkan parameter FileID
        requiredInputs = []
        for i in range(GlobalParams().NumberBooks):
            requiredInputs.append(CountWords(FileID=i))
        return requiredInputs
#file output untuk output data kata terpopuler
def output(self):
        return luigi.LocalTarget("data/summary.txt")
#menggunakan pickle konversi data menjadi python object
def run(self):
#counter object sebagai penyimpanan total penghitungan
        total_count = Counter()
        for input in self.input():
            with input.open("rb") as infile:
                nextCounter = pickle.load(infile)
                total_count += nextCounter
#iterasi aggregasi penghitungan kata dari setiap file, dan output hasil
with self.output().open("w") as f:
            for item in total_count.most_common(GlobalParams().NumberTopWords):
                f.write("{0: <15}{1}\n".format(*item))

Eksekusi task, penulisan setiap task dengan penambahan parameter, batasan buku yang akan dianalisis dan batasan kata yang akan dimasukkan kedalam data output.

python -m luigi --module extract-link TopWords --GlobalParams-NumberBooks 15 --GlobalParams-NumberTopWords 750

Pada center scheduler interface akan tampil grafik flow dari luigi task yang telah dieksekusi.

Pada folder data, akan muncul file output berupa kata-kata terpopuler dari data buku yang telah dianalisis. Sesuai pada task TopWords nama file yaitu summary.txt.

Kesimpulan

Luigi dari python dapat digunakan untuk membuat data pipeline, menggunakan task, parameter, dan scheduler.

Dengan memahami konsep luigi, maka framework ini cukup powerful untuk mengekstraksi data dari beberapa sumber data, cukup menambahkan code untuk ekstraksi data pada task, juga bisa digunakan untuk data yang cukup besar dan kompleks, lebih lengkapnya tentang konsep luigi dapat dibaca pada dokumentasi luigi.