5 Alternatif Airflow Teratas untuk Orkestrasi Data (Disertai Contoh Kode)

Baca blog Airflow vs Prefect: Memutuskan Mana yang Tepat untuk Workflow Data Anda untuk mempelajari perbandingan mendetail antara dua alat orkestrasi data ini. 

Mulai menggunakan Prefect

Kita akan memulai proyek Prefect dengan memasang paket Python. Jalankan perintah berikut di terminal.

$ pip install -U prefect

Setelah itu, kita akan membuat skrip Python bernama prefect_etl.py dan menuliskan kode berikut.

from prefect import task, flow
import pandas as pd

# Extract data
@task
def extract_data():
    # Simulating data extraction
    data = {
        "name": ["Alice", "Bob", "Charlie"],
        "age": [25, 30, 35],
        "city": ["New York", "Los Angeles", "Chicago"]
    }
    df = pd.DataFrame(data)
    return df

# Transform data
@task
def transform_data(df: pd.DataFrame):
    # Example transformation: adding a new column
    df["age_plus_ten"] = df["age"] + 10
    return df

# Load data
@task
def load_data(df: pd.DataFrame):
    # Simulating data load
    print("Loading data to target destination:")
    print(df)

# Defining the flow
@flow(log_prints=True)
def etl():
    raw_data = extract_data()
    transformed_data = transform_data(raw_data)
    load_data(transformed_data)

# Running the flow
if __name__ == "__main__":
    etl()

Kode di atas mendefinisikan fungsi tugas extract_data(), transform_data(), dan load_data() lalu mengeksekusinya secara berurutan dalam fungsi flow bernama etl(). Fungsi-fungsi ini dibuat menggunakan dekorator Python Prefect. 

Singkatnya, kita membuat DataFrame pandas, mentransformasikannya, lalu menampilkan hasil akhir menggunakan print. Ini cara sederhana untuk mensimulasikan pipeline ETL.

Untuk mengeksekusi workflow, jalankan saja skrip Python menggunakan perintah berikut.

$ python prefect_etl.py 

Seperti yang terlihat, run workflow kita berhasil diselesaikan.

Log run flow Prefect.

Mendeploy flow

Sekarang kita akan mendeploy workflow agar bisa dijalankan terjadwal atau dipicu berdasarkan peristiwa. Deployment flow juga memungkinkan kita memantau dan mengelola banyak workflow secara terpusat.

Untuk mendeploy flow, kita akan menggunakan Prefect CLI. Fungsi deploy memerlukan nama file Python, nama fungsi flow di file tersebut, dan nama deployment. Dalam contoh ini, kita menamai deployment ini “simple_etl.”

$ prefect deploy prefect_etl.py:etl -n 'simple_etl'

Setelah menjalankan skrip di atas di terminal, Anda mungkin menerima pesan bahwa Anda belum memiliki worker pool untuk menjalankan deployment. Untuk membuat worker pool, gunakan perintah berikut.

$ prefect worker start --pool 'datacamp'

Sekarang kita sudah memiliki worker pool, buka jendela terminal lain dan jalankan deployment. Perintah prefect deployment run memerlukan argumen “<flow-function-name>/<deployment-name>”, seperti pada perintah di bawah.

$ prefect deployment run 'etl/simple_etl

Sebagai hasil menjalankan deployment, Anda akan menerima pesan bahwa workflow sedang berjalan. Biasanya, flow run yang dibuat akan diberi nama acak, dalam kasus saya witty-lorikeet.

Creating flow run for deployment 'etl/simple_etl'...
Created flow run 'witty-lorikeet'.
└── UUID: 4e0495b0-9c7e-4ed8-b9ab-5160994dc7f0
└── Parameters: {}
└── Job Variables: {}
└── Scheduled start time: 2024-06-22 14:05:01 PKT (now)
└── URL: <no dashboard available>

Untuk melihat log lengkap, kembali ke jendela terminal tempat Anda memulai workers pool.

Ringkasan run flow Prefect.

Anda harus memulai web server Prefect untuk memvisualisasikan run flow secara lebih ramah pengguna dan mengelola workflow lainnya.

$ prefect server start 

Setelah mengeksekusi perintah di atas, Anda akan diarahkan ke dasbor Prefect. Atau, Anda bisa langsung membuka alamat http://127.0.0.1:4200 di peramban Anda.

UI web server Prefect

Dasbor memungkinkan Anda menjalankan ulang workflow, melihat log, memeriksa work pool, mengatur notifikasi, dan memilih opsi lanjutan lainnya. Ini adalah solusi lengkap untuk kebutuhan orkestrasi data modern Anda.

Untuk mempelajari cara membangun dan mengeksekusi pipeline machine learning menggunakan Prefect, Anda dapat mengikuti tutorial Using Prefect for Machine Learning Workflows.

2. Dagster

Dasgter adalah framework open-source yang dirancang bagi engineer data untuk mendefinisikan, menjadwalkan, dan memantau pipeline data. Framework ini sangat skalabel dan memfasilitasi kolaborasi antar berbagai tim data. 

Dagster memungkinkan pengguna mendefinisikan aset data mereka sebagai fungsi Python menggunakan dekorator. Setelah aset ini didefinisikan, pengguna dapat mengeksekusinya dengan mulus melalui penjadwalan atau pemicu berbasis peristiwa.

Dibandingkan dengan Airflow, Dagster memungkinkan kita mengembangkan, menguji, dan meninjau pipeline secara lokal, menyediakan pendekatan orkestrasi berbasis aset, serta native terhadap cloud dan container.

Mulai menggunakan Dagster

Kita akan membuat pipeline ETL sederhana, mengeksekusinya, dan memvisualisasikannya menggunakan web server Dagster. Mirip dengan dasbor Prefect, web server Dagster menyediakan cara terpusat untuk memantau banyak workflow serta menjadwalkan run dan aset.

Kita akan mulai dengan memasang paket Python.

$ pip install dagster -q

Lalu, kita akan membuat tiga fungsi Python untuk mengekstrak, mentransformasi, dan memuat data. Fungsi-fungsi ini diberi nama create_dirty_data(), clean_data(), dan load_cleaned_data() pada kode. Dengan dekorator @asset, kita akan mendeklarasikan fungsi-fungsi tersebut sebagai aset data di Dagster.

Selanjutnya, kita akan membuat asset job (variabel job) menggunakan semua aset (variabel all_assets) lalu membuat definisi aset (variabel defs). 

Anda bisa melewatkan bagian definisi aset, namun ini menjadi penting jika Anda ingin menjadwalkan run, menjalankan beberapa job, dan menyiapkan sensor.

import pandas as pd
import numpy as np
from dagster import asset, Definitions, define_asset_job, materialize

@asset
def create_dirty_data():
    # Create a sample DataFrame with dirty data
    data = {
        'Name': [' John Doe ', 'Jane Smith', 'Bob Johnson ', '  Alice Brown'],
        'Age': [30, np.nan, 40, 35],
        'City': ['New York', 'los angeles', 'CHICAGO', 'Houston'],
        'Salary': ['50,000', '60000', '75,000', 'invalid']
    }
    df = pd.DataFrame(data)
    
    # Save the DataFrame to a CSV file
    dirty_file_path = 'dag_data/dirty_data.csv'
    df.to_csv(dirty_file_path, index=False)
    
    return dirty_file_path

@asset
def clean_data(create_dirty_data):
    # Read the dirty CSV file
    df = pd.read_csv(create_dirty_data)
    
    # Clean the data
    df['Name'] = df['Name'].str.strip()
    df['Age'] = pd.to_numeric(df['Age'], errors='coerce').fillna(df['Age'].mean())
    df['City'] = df['City'].str.upper()
    df['Salary'] = df['Salary'].replace('[\$,]', '', regex=True)
    df['Salary'] = pd.to_numeric(df['Salary'], errors='coerce').fillna(0)
    
    # Calculate average salary
    avg_salary = df['Salary'].mean()
    
    # Save the cleaned DataFrame to a new CSV file
    cleaned_file_path = 'dag_data/cleaned_data.csv'
    df.to_csv(cleaned_file_path, index=False)
    
    return {
        'cleaned_file_path': cleaned_file_path,
        'avg_salary': avg_salary
    }

@asset
def load_cleaned_data(clean_data):
    cleaned_file_path = clean_data['cleaned_file_path']
    avg_salary = clean_data['avg_salary']
    
    # Read the cleaned CSV file to verify
    df = pd.read_csv(cleaned_file_path)
    
    print({
        'num_rows': len(df),
        'num_columns': len(df.columns),
        'avg_salary': avg_salary
    })

# Define all assets
all_assets = [create_dirty_data, clean_data, load_cleaned_data]

# Create a job that will materialize all assets
job = define_asset_job("all_assets_job", selection=all_assets)

# Create Definitions object
defs = Definitions(
    assets=all_assets,
    jobs=[job]
)

if __name__ == "__main__":
    result = materialize(all_assets)
    print("Pipeline execution result:", result.success)

Anda dapat menjalankan kode di atas di Jupyter Notebook atau membuat file Python dan menjalankannya. 

Sebagai hasil eksekusi kode, kita akan mendapatkan log lengkap dari run workflow. 

Web server Dagster

Untuk memvisualisasikan aset dan run job, kita harus memasang dan menjalankan web server Dagster. Web server memungkinkan Anda menjalankan job, mematerialisasi aset individual, dan memantau banyak job sekaligus.

$ pip install dagster-webserver

Untuk memulai server Dagster, kita akan menggunakan Dagster CLI dan memberinya lokasi file Python. Dalam kasus ini, saya menamai filenya dagster_pipe.py.

$ dagster dev -f dagster_pipe.py  

Perintah di atas akan meluncurkan web server di peramban Anda secara otomatis. Sebagai alternatif, Anda bisa langsung membuka alamat http://127.0.0.1:3000 di peramban Anda.

UI web server Dagster.

Sejauh ini kita baru mendeploy job. Untuk menjalankan workflow, buka tab “Runs” dan klik tombol “Launch a new run”. 

Run seharusnya berhasil! Untuk melihat log, klik ID run yang Anda minati.

Log run Dagster.

3. Mage AI

Mage AI adalah framework orkestrasi data hibrida open-source. Hibrida berarti Anda mendapatkan fleksibilitas Jupyter Notebook dan kendali dari kode modular. 

Siapa pun, bahkan dengan pengetahuan Python terbatas, dapat membangun, menjalankan, dan memantau pipeline data. Alih-alih menulis dan menjalankan file Python secara langsung, Anda akan membuat proyek Mage AI dan meluncurkannya di dasbor, tempat Anda bisa membangun, menjalankan, dan mengelola pipeline data.

Dibandingkan Airflow, Mage AI menyediakan antarmuka yang ramah pengguna dan mudah digunakan, menjadikannya pilihan tepat bagi mereka yang baru di bidang rekayasa data. Alat ini dirancang dengan skalabilitas dan mampu menangani volume data besar serta struktur pipeline kompleks secara efisien.

Mulai menggunakan Mage AI

Memulai Mage AI cukup sederhana. Kita hanya perlu memasang paket Python Mage AI.

$ pip install mage-ai

Dan memulai proyek Mage AI. 

$ mage start mage_ai_etl 

Perintah di atas akan memulai web server. Seperti disebutkan sebelumnya, semua pengeditan kode, menjalankan job, dan memantau job dilakukan melalui UI Mage AI.

UI Mage AI.

Klik “+ New pipeline” untuk membuat pipeline ETL pertama Anda. Saya menamainya “simple_etl.”

Membuat pipeline baru di Mage AI.

Kemudian, antarmuka akan meminta Anda menambahkan modul untuk mulai menulis kode. Pilih modul “Data Loader” dan tulis kode Python berikut. 

Di sini, kita mendeklarasikan fungsi create_sample_csv() yang menjadi langkah pertama pada pipeline kita. Kita menggunakan dekorator @data_loader dari Mage AI. Kita juga mendefinisikan fungsi test_output() yang memastikan keluaran ada. Ini membantu manajemen dependensi tugas.

import io
import pandas as pd

if 'data_loader' not in globals():
    from mage_ai.data_preparation.decorators import data_loader
if 'test' not in globals():
    from mage_ai.data_preparation.decorators import test

@data_loader
def create_sample_csv() -> pd.DataFrame:
    """
    Create a sample CSV file with duplicates and missing values
    """
    csv_data = """
category,product,quantity,price
Electronics,Laptop,5,1000
Electronics,Smartphone,10,500
Clothing,T-shirt,50,20
Clothing,Jeans,30,50
Books,Novel,100,15
Books,Textbook,20,80
Electronics,Laptop,5,1000
Clothing,T-shirt,,20
Electronics,Tablet,,300
Books,Magazine,25,
"""
    return pd.read_csv(io.StringIO(csv_data.strip()))
   
@test
def test_output(df) -> None:
    """
    Template code for testing the output of the block.
    """
    assert df is not None, 'The output is undefined'

Membuat blok data loader di Mage AI.

Selanjutnya, buat modul lain bernama “Transformer” dan tambahkan fungsi clean_data() seperti pada kode di bawah. 

Anda boleh mengabaikan fungsi test(); Anda cukup menambahkan fungsi transformer utama, yaitu clean_data().

import pandas as pd

if 'transformer' not in globals():
    from mage_ai.data_preparation.decorators import transformer
if 'test' not in globals():
    from mage_ai.data_preparation.decorators import test

@transformer
def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    Clean and transform the data
    """
    # Remove duplicates
    df = df.drop_duplicates()
    # Fill missing values with 0
    df = df.fillna(0)
    return df

@test
def test_output(df) -> None:
    """
    Template code for testing the output of the block.
    """
    assert df is not None, 'The output is undefined'

Serupa, buat modul “Data Exporter” dan tambahkan kode berikut. Kode ini mendeklarasikan fungsi pemuatan data, export_data_to_csv(), yang menyimpan data yang telah ditransformasi ke file CSV. 

import pandas as pd

if 'data_exporter' not in globals():
    from mage_ai.data_preparation.decorators import data_exporter

@data_exporter
def export_data_to_csv(df: pd.DataFrame) -> None:
    """
    Export the processed data to a CSV file
    """
    df.to_csv('output_data.csv', index=False)
    print("Data exported successfully to output_data.csv")

Untuk menjalankan pipeline, buka tab “Trigger”, dan klik “Run@once”.

Menjalankan pipeline di Mage AI.

Untuk melihat log run, buka tab “Runs” dan klik tombol “Logs” pada pipeline yang baru dijalankan.

Log run flow Mage AI.

4. Kedro

Kedro adalah framework orkestrasi data open-source populer lainnya yang sedikit berbeda dari alat lain. Alat ini dibuat untuk engineer machine learning dan mengadopsi banyak konsep dari rekayasa perangkat lunak, menerapkannya pada proyek machine learning.

Kedro dirancang sangat modular, yang berarti bahkan untuk mengekspor dataset, Anda harus membuat katalog data yang menentukan lokasi dan jenis data, memastikan manajemen data yang standar dan efisien di seluruh pipeline.

Untuk memahami bagaimana Kedro cocok dalam ekosistem machine learning, Anda dapat mengeksplorasi berbagai alat MLOps dengan membaca artikel 25 Top MLOps Tools You Need to Know in 2024.

Dibandingkan Airflow, API Kedro lebih sederhana untuk membangun pipeline data. Ia lebih berfokus pada rekayasa machine learning dan menawarkan kategorisasi serta versi data.

Mulai menggunakan Kedro

Membangun pipeline data Kedro adalah cerita yang berbeda. Framework ini modular, dan Anda perlu memahami struktur proyek dan berbagai langkah yang terlibat untuk mengeksekusi workflow dengan sukses. 

Mulailah dengan memasang paket Python Kedro. 

$ pip install kedro

Inisialisasi proyek Kedro. 

$ kedro new --name=kedro_etl --tools=none --example=n 

Pindah ke direktori proyek. 

$ cd kedro-etl  

Buat folder di dalam folder pipelines bernama data_processing.

$ mkdir -p src/kedro_etl/pipelines/data_processing  

Buat file Python bernama kedro_pipe.py dan buka di IDE favorit Anda, misalnya Visual Studio Code.

$ code src/kedro_etl/pipelines/data_processing/kedro_pipe.py

Skrip Python harus memuat fungsi extract, transform, dan load, yang merupakan node dalam pipeline. Dalam hal ini, yaitu fungsi create_sample_data(), clean_data(), dan load_and_process_data().

Lalu, kita merangkai node-node tersebut menggunakan kelas Kedro Pipeline di dalam fungsi create_pipeline(). Di fungsi pipeline, kita mendefinisikan node, dan setiap node memiliki inputs, outputs, dan name node. 

import pandas as pd
import numpy as np
from kedro.pipeline import Pipeline, node

def create_sample_data():
    data = {
        'id': range(1, 101),
        'name': [f'Person_{i}' for i in range(1, 101)],
        'age': np.random.randint(18, 80, 100),
        'salary': np.random.randint(20000, 100000, 100),
        'missing_values': [np.nan if i % 10 == 0 else i for i in range(100)]
    }
    return pd.DataFrame(data)

def clean_data(df: pd.DataFrame):
    # Remove rows with missing values
    df_cleaned = df.dropna()

    # Convert salary to thousands
    df_cleaned['salary'] = df_cleaned['salary'] / 1000

    # Capitalize names
    df_cleaned['name'] = df_cleaned['name'].str.upper()

    return df_cleaned

def load_and_process_data(df: pd.DataFrame):
    # Calculate average salary
    avg_salary = df['salary'].mean()

    # Add a new column for salary category
    df['salary_category'] = df['salary'].apply(
        lambda x: 'High' if x > avg_salary else 'Low')

    # Calculate age groups
    df['age_group'] = pd.cut(df['age'], bins=[0, 30, 50, 100], labels=[
                             'Young', 'Middle', 'Senior'])

    print(df)
    return df

def create_pipeline(**kwargs):
    return Pipeline(
        [
            node(
                func=create_sample_data,
                inputs=None,
                outputs="raw_data",
                name="create_sample_data_node",
            ),
            node(
                func=clean_data,
                inputs="raw_data",
                outputs="cleaned_data",
                name="clean_data_node",
            ),
            node(
                func=load_and_process_data,
                inputs="cleaned_data",
                outputs="processed_data",
                name="load_and_process_data_node",
            ),
        ]
    )

Jika kita menjalankan pipeline tanpa membuat katalog data, data kita tidak akan diekspor. Jadi, kita perlu membuka file conf/base/catalog.yml dan mengeditnya dengan menyediakan konfigurasi dataset.

raw_data:
  type: pandas.CSVDataset
  filepath: ./data/kedro/sample_data.csv

cleaned_data:
  type: pandas.CSVDataset
  filepath: ./data/kedro/cleaned_data.csv

processed_data:
  type: pandas.CSVDataset
  filepath: ./data/kedro/processed_data.csv

Kita juga harus memasukkan file Python yang baru dibuat ke dalam registry pipeline. Untuk melakukannya, buka file Python src/simple_etl/pipeline_registry.py dan tambahkan kode berikut. 

"""Project pipelines."""
from __future__ import annotations

from kedro.pipeline import Pipeline
from kedro_etl.pipelines.data_processing import kedro_pipe

def register_pipelines() -> Dict[str, Pipeline]:
    data_processing_pipeline = kedro_pipe.create_pipeline()

    return {
        "__default__": data_processing_pipeline,
        "data_processing": data_processing_pipeline,
    }

Jalankan pipeline dan lihat log langsung di terminal dengan menjalankan perintah berikut.

$ kedro run

Log run pipeline Kedro.

Setelah menjalankan pipeline, file Anda akan disimpan dalam format CSV di lokasi yang ditentukan di katalog data.

File keluaran run pipeline Kedro.

Jika Anda mengalami masalah saat menjalankan pipeline, pertimbangkan memasang Kedro dengan semua ekstensi. 

$ pip install "kedro[all]"

Visualisasi Kedro

Kita dapat memvisualisasikan dan membagikan pipeline dengan memasang alat kedro-viz. 

$ pip install kedro-viz

Kemudian, mengeksekusi perintah berikut memungkinkan kita memvisualisasikan seluruh pipeline data dan node data. Ini juga menyediakan opsi pelacakan eksperimen dan kemampuan berbagi visualisasi pipeline.

$ kedro viz run

Visualisasi pipeline Kedro.

5. Luigi

Luigi adalah framework berbasis Python open-source yang dikembangkan oleh Spotify dan unggul dalam mengelola proses batch jangka panjang serta pipeline data kompleks. Alat ini andal dalam resolusi dependensi, manajemen workflow, visualisasi, dan pemulihan kegagalan, sehingga menjadi alat yang kuat untuk mengorkestrasi workflow data. 

Dibandingkan Airflow, Luigi memiliki API minimalis, penjadwalan kalender, dan basis pengguna setia yang akan membantu Anda dengan masalah terkait pipeline orkestrasi data. 

Mulai menggunakan Luigi

Membuat pipeline data Luigi memerlukan pemahaman pemrograman berorientasi objek. Mari mulai dengan memasang paket Python Luigi. 

$ pip install luigi

Untuk mengembangkan pipeline ETL sederhana di Luigi, kita akan membuat tugas-tugas yang saling terhubung. Alih-alih membuat fungsi Python sebagai tugas, kita akan membuat kelas Python untuk setiap langkah dalam pipeline, yaitu FetchData, ProcessData dan GenerateReport. Setiap kelas akan memiliki tiga fungsi: requires(), output(), dan run(). 

Fungsi requires() dan output() akan menghubungkan tugas, dan fungsi run() akan mengeksekusi kode pemrosesan. Di akhir, kita akan membangun pipeline menggunakan tugas terakhir pada pipeline. 

import luigi
import pandas as pd
import numpy as np

class FetchData(luigi.Task):
    def output(self):
        return luigi.LocalTarget('data/fetch_data.csv')
    
    def run(self):
        # Simulate fetching data by creating a sample CSV file
        data = {
            'column1': [1, 2, np.nan, 4],
            'column2': ['A', 'B', 'C', np.nan]
        }
        df = pd.DataFrame(data)
        df.to_csv(self.output().path, index=False)

class ProcessData(luigi.Task):
    def requires(self):
        return FetchData()
    
    def output(self):
        return luigi.LocalTarget('data/process_data.csv')
    
    def run(self):
        df = pd.read_csv(self.input().path)
        # Fill missing values
        df['column1'].fillna(df['column1'].mean(), inplace=True)
        df['column2'].fillna('B', inplace=True)
        df.to_csv(self.output().path, index=False)

class GenerateReport(luigi.Task):
    def requires(self):
        return ProcessData()
    
    def output(self):
        return luigi.LocalTarget('data/generate_report.txt')
    
    def run(self):
        df = pd.read_csv(self.input().path)
        # Simple data analysis: calculate mean of column1 and value counts of column2
        mean_column1 = df['column1'].mean()
        value_counts_column2 = df['column2'].value_counts()
        
        with self.output().open('w') as out_file:
            out_file.write(f'Mean of column1: {mean_column1}\n')
            out_file.write('Value counts of column2:\n')
            out_file.write(value_counts_column2.to_string())

if __name__ == '__main__':
    luigi.build([GenerateReport()], local_scheduler=True)

Jalankan kode di atas di Jupyter Notebook atau buat file Python dan jalankan melalui terminal. 

Serupa dengan Luigi, Anda juga dapat mempelajari cara membangun pipeline ETL dengan Apache Airflow. Tutorial tersebut membahas dasar-dasar mengekstrak, mentransformasi, dan memuat data dengan Apache Airflow.

Central planner Luigi

Kita perlu menginisialisasi central planner Luigi untuk menjadwalkan run pipeline atau memicunya dengan suatu peristiwa.

Mulai scheduler dengan mengetik perintah berikut di terminal.

$ luigid  

2024-06-22 13:35:18,636 luigi[25056] INFO: logging configured by default settings
2024-06-22 13:35:18,636 luigi.scheduler[25056] INFO: No prior state file exists at /var/lib/luigi-server/state.pickle. Starting with empty state
2024-06-22 13:35:18,640 luigi.server[25056] INFO: Scheduler starting up

Untuk menjalankan pipeline, buka terminal baru dan ketik perintah berikut. Perintah Luigi memerlukan nama file Python dan tugas terakhir yang ingin kita eksekusi. Dalam kasus ini, nama file adalah luigi_pipe.py, dan tugas Luigi terakhir kita adalah GenerateReport.

$ python -m luigi --module luigi_pipe GenerateReport

Jika Anda ingin memvisualisasikan run pipeline dan status tugas, Anda bisa langsung membuka http://localhost:8082 di peramban Anda.

webUI Luigi Central Planner.

Demikian panduan 5 alternatif terbaik untuk Airflow! Jika Anda ingin mendalami salah satu contoh yang disajikan dalam artikel ini, berikut beberapa sumber yang bisa dipertimbangkan:

• Untuk kode sumber dan data Prefect, Dagster, dan Luigi, silakan merujuk ke ruang kerja DataLab.
• Untuk kode sumber dan data Mage AI dan Kedro, silakan merujuk ke repositori GitHub.

Penutup

Dalam tutorial ini, kita telah membahas alternatif Airflow open-source dan gratis terbaik. Kita juga telah mempelajari masing-masing alat orkestrasi data, serta membangun dan mengeksekusi pipeline ETL sederhana. Melihat contoh kode akan membantu Anda memutuskan mana yang paling cocok untuk kasus penggunaan Anda.

Jika Anda pemula, saya sarankan memulai dengan Prefect atau Mage AI karena keduanya ramah pengguna dan mudah disiapkan. Namun, jika Anda mencari alat yang lebih canggih dan mengikuti praktik rekayasa perangkat lunak, saya rekomendasikan mengeksplorasi Dagster, Kedro, dan Luigi.

Setelah mengeksplorasi artikel ini, langkah alami berikutnya dalam perjalanan rekayasa data Anda adalah mendapatkan sertifikasi seperti Data Engineer in Python dari DataCamp untuk mempelajari alat lainnya dan membangun pipeline data end-to-end yang bisa Anda deploy ke produksi.