Tutorial Flyte: Workflow Orchestration untuk Machine Learning dan Data Engineering

Flyte adalah platform workflow orchestration open-source yang dirancang khusus untuk machine learning, data engineering, dan data science. Dikembangkan oleh Union.ai dan digunakan oleh perusahaan seperti Spotify, Lyft, dan Freenome, Flyte menyediakan cara yang scalable, reproducible, dan production-ready untuk mengelola pipeline data dan ML.

Berbeda dengan workflow orchestrator tradisional seperti Airflow yang fokus pada scheduling task, Flyte dibangun dari awal dengan pemahaman mendalam tentang kebutuhan ML: versioning data dan model, type safety, caching hasil komputasi, dan native support untuk container execution. Dalam tutorial ini, kita akan mempelajari Flyte dari instalasi hingga membangun pipeline ML yang siap production.

Mengapa Flyte?

Sebelum memulai, mari pahami mengapa Flyte menjadi pilihan yang semakin populer di ekosistem MLOps:

Instalasi dan Setup

Prasyarat

Pastikan sistem Anda memiliki:

• Python 3.8 atau lebih baru
• Docker (untuk menjalankan Flyte secara lokal)
• pip atau conda untuk manajemen paket

Instalasi Flytekit

Flytekit adalah SDK Python untuk menulis workflow Flyte:

pip install flytekit

Untuk fitur tambahan seperti integrasi Pandas dan visualisasi:

pip install flytekit[pandas]
pip install flytekitplugins-deck-standard

Menjalankan Flyte Secara Lokal

Flyte menyediakan sandbox untuk development lokal menggunakan Docker:

pip install flytectl
flytectl demo start

Perintah ini akan menjalankan cluster Flyte lengkap di Docker, termasuk:

• Flyte Admin (API server)
• Flyte Console (web UI)
• Flyte Propeller (workflow engine)
• MinIO (object storage)
• PostgreSQL (metadata store)

Setelah berhasil, akses Flyte Console di http://localhost:30080/console.

Alternatif yang lebih ringan, Anda bisa menjalankan workflow secara lokal tanpa cluster:

pyflyte run myworkflow.py myworkflow --inputparam value

from flytekit import task

@task
def sayhello(name: str) -> str:
    return f"Hello, {name}!"

Perhatikan bahwa type hint wajib digunakan. Flyte menggunakan type hint untuk validasi dan serialisasi data.

Workflows

Workflow menggabungkan beberapa task menjadi sebuah pipeline:

from flytekit import task, workflow

@task
def sayhello(name: str) -> str:

    return f"Hello, {name}!"

@task
def greetinglength(greeting: str) -> int:
    return len(greeting)

@workflow
def myworkflow(name: str) -> int:

    greeting = sayhello(name=name)
    length = greetinglength(greeting=greeting)

    return length

# Jalankan secara lokal (untuk development)
pyflyte run myworkflow.py myworkflow --name "Flyte"


Jalankan di cluster Flyte
pyflyte run --remote myworkflow.py myworkflow --name "Flyte"

if name == "main":
    result = myworkflow(name="Flyte")
    print(f"Result: {result}")

Penggunaan Dasar: Pipeline Data Processing

Mari bangun pipeline data processing sederhana menggunakan Flyte:

import pandas as pd
from flytekit import task, workflow
from flytekit.types.file import FlyteFile

@task
def loaddata(filepath: str) -> pd.DataFrame:
    """Membaca dataset dari file CSV."""
    df = pd.readcsv(filepath)
    print(f"Loaded {len(df)} rows")
    return df

@task
def cleandata(df: pd.DataFrame) -> pd.DataFrame:

    """Membersihkan data: hapus duplikat dan missing values."""
    initialrows = len(df)
    df = df.dropduplicates()

    df = df.dropna()
    cleanedrows = len(df)
    print(f"Cleaned: {initialrows} -> {cleanedrows} rows")
    return df

@task
def computestatistics(df: pd.DataFrame) -> dict:

    """Menghitung statistik dasar dari dataset."""
    stats = {
        "rowcount": len(df),
        "columncount": len(df.columns),

        "columns": list(df.columns),
        "numericmeans": df.selectdtypes(include="number").mean().todict(),
    }
    return stats

@workflow
def datapipeline(filepath: str) -> dict:
    """Pipeline lengkap: load, clean, dan compute statistics."""
    rawdata = loaddata(filepath=filepath)
    clean = cleandata(df=rawdata)
    stats = computestatistics(df=clean)

    return stats

pyflyte run datapipeline.py datapipeline --filepath "./data/sales.csv"

Penggunaan Lanjutan: ML Training Pipeline

Pipeline Training Model dengan Caching

Salah satu fitur terbaik Flyte adalah caching. Dengan menambahkan cache=True, task yang input-nya tidak berubah tidak akan dijalankan ulang:

import pandas as pd
import numpy as np
from flytekit import task, workflow, Resources
from flytekit.types.file import FlyteFile
from sklearn.modelselection import traintestsplit

from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracyscore, f1score

import joblib
import os
from dataclasses import dataclass
from mashumaro.mixins.json import DataClassJSONMixin

@dataclass
class TrainingConfig(DataClassJSONMixin):
    testsize: float = 0.2
    randomstate: int = 42

    nestimators: int = 100
    maxdepth: int = 10


@dataclass
class ModelMetrics(DataClassJSONMixin):
    accuracy: float
    f1score: float
    featureimportances: dict


@task(cache=True, cacheversion="1.0")
def preparedataset(

    filepath: str, config: TrainingConfig
) -> tuple[pd.DataFrame, pd.DataFrame, pd.Series, pd.Series]:
    """Load dan split dataset menjadi train/test."""
    df = pd.readcsv(filepath)

    X = df.drop(columns=["target"])
    y = df["target"]

    Xtrain, Xtest, ytrain, ytest = traintestsplit(
        X, y, testsize=config.testsize, randomstate=config.randomstate
    )

    return Xtrain, Xtest, ytrain, ytest

@task(
    requests=Resources(cpu="2", mem="4Gi"),
    limits=Resources(cpu="4", mem="8Gi"),
)
def trainmodel(

    Xtrain: pd.DataFrame,
    ytrain: pd.Series,

    config: TrainingConfig,
) -> FlyteFile:
    """Training model RandomForest dan simpan sebagai file."""
    model = RandomForestClassifier(
        nestimators=config.nestimators,

        maxdepth=config.maxdepth,

        randomstate=config.randomstate,

        njobs=-1,
    )
    model.fit(Xtrain, ytrain)

    modelpath = "/tmp/model.joblib"

    joblib.dump(model, modelpath)

    return FlyteFile(path=modelpath)


@task
def evaluatemodel(
    modelfile: FlyteFile,

    Xtest: pd.DataFrame,
    ytest: pd.Series,

    featurenames: list[str],
) -> ModelMetrics:
    """Evaluasi model dan kembalikan metrik."""
    modelfile.download()

    model = joblib.load(modelfile.path)

    predictions = model.predict(Xtest)


    importances = dict(
        zip(featurenames, model.featureimportances.tolist())
    )

    return ModelMetrics(
        accuracy=accuracyscore(ytest, predictions),
        f1score=f1score(ytest, predictions, average="weighted"),

        featureimportances=importances,
    )

@workflow
def mltrainingpipeline(
    filepath: str, config: TrainingConfig = TrainingConfig()

) -> ModelMetrics:
    """Pipeline ML lengkap: prepare, train, evaluate."""
    Xtrain, Xtest, ytrain, ytest = preparedataset(
        filepath=filepath, config=config
    )
    modelfile = trainmodel(
        Xtrain=Xtrain, ytrain=ytrain, config=config
    )
    metrics = evaluatemodel(

        modelfile=modelfile,

        Xtest=Xtest,

        ytest=ytest,

        featurenames=list(Xtrain.columns),

    )
    return metrics

from flytekit import task, workflow, dynamic

@task
def trainsinglemodel(

    Xtrain: pd.DataFrame,
    ytrain: pd.Series,

    nestimators: int,
) -> float:
    """Train satu model dan kembalikan accuracy."""
    model = RandomForestClassifier(nestimators=nestimators, randomstate=42)

    model.fit(Xtrain, ytrain)

    return model.score(Xtrain, ytrain)


@dynamic
def hyperparametersearch(
    Xtrain: pd.DataFrame,

    ytrain: pd.Series,
    nestimatorsoptions: list[int],
) -> list[float]:
    """Jalankan training untuk setiap kombinasi hyperparameter."""
    results = []
    for nest in nestimatorsoptions:

        score = trainsinglemodel(

            Xtrain=Xtrain,

            ytrain=ytrain,

            nestimators=nest,

        )
        results.append(score)
    return results

@workflow
def searchpipeline(
    filepath: str,

) -> list[float]:
    Xtrain, Xtest, ytrain, ytest = preparedataset(
        filepath=filepath, config=TrainingConfig()
    )
    scores = hyperparametersearch(

        Xtrain=Xtrain,

        ytrain=ytrain,

        nestimatorsoptions=[50, 100, 200, 500],

    )
    return scores

from flytekit import task, workflow, conditional

@task
def calculateaccuracy(modelfile: FlyteFile, Xtest: pd.DataFrame, ytest: pd.Series) -> float:

    modelfile.download()
    model = joblib.load(modelfile.path)

    return model.score(Xtest, ytest)


@task
def deploymodel(modelfile: FlyteFile) -> str:

    return f"Model deployed: {modelfile.path}"

@task
def retrainnotification() -> str:

    return "Model accuracy below threshold. Retraining needed."

@workflow
def deploymentpipeline(
    filepath: str, accuracythreshold: float = 0.85
) -> str:
    Xtrain, Xtest, ytrain, ytest = preparedataset(

        filepath=filepath, config=TrainingConfig()

    )
    modelfile = trainmodel(

        Xtrain=Xtrain, ytrain=ytrain, config=TrainingConfig()

    )
    accuracy = calculateaccuracy(
        modelfile=modelfile, Xtest=Xtest, ytest=ytest
    )
    result = (
        conditional("deploymentcheck")

        .if(accuracy >= accuracythreshold)

        .then(deploymodel(modelfile=modelfile))
        .else()

        .then(retrainnotification())
    )
    return result

Map Tasks untuk Parallel Execution

Untuk menjalankan task yang sama dengan input berbeda secara paralel:

from flytekit import task, workflow, maptask

@task
def processpartition(partitionid: int) -> dict:

    """Proses satu partisi data."""
    return {
        "partitionid": partitionid,

        "rowsprocessed": partitionid  1000,

        "status": "completed",
    }

@task
def aggregateresults(results: list[dict]) -> dict:
    """Gabungkan hasil dari semua partisi."""
    totalrows = sum(r["rowsprocessed"] for r in results)
    return {
        "totalpartitions": len(results),

        "totalrows": totalrows,

    }

@workflow
def parallelpipeline(numpartitions: int = 10) -> dict:

    partitionids = list(range(numpartitions))

    results = maptask(processpartition)(partitionid=partitionids)

    summary = aggregateresults(results=results)
    return summary

import mlflow
from flytekit import task, workflow
from flytekitplugins.mlflow import mlflowautolog


@task(enabledeck=True)
@mlflowautolog(framework=mlflow.sklearn)

def trainwithtracking(

    Xtrain: pd.DataFrame,
    ytrain: pd.Series,

    nestimators: int,
) -> FlyteFile:
    """Training dengan MLflow autologging."""
    model = RandomForestClassifier(nestimators=nestimators)
    model.fit(Xtrain, ytrain)

    modelpath = "/tmp/model.joblib"

    joblib.dump(model, modelpath)
    return FlyteFile(path=modelpath)

from flytekit import task
import greatexpectations as gx

@task
def validatedata(df: pd.DataFrame) -> bool:

    """Validasi kualitas data sebelum training."""
    context = gx.getcontext()

    datasource = context.datasources.addpandas("pandasds")

    dataasset = datasource.adddataframeasset("myasset")

    batch = dataasset.addbatchdefinitionwholedataframe("batch").getbatch(

        batchparameters={"dataframe": df}
    )

    expectationsuite = context.suites.add(

        gx.ExpectationSuite(name="trainingdatasuite")

    )
    expectationsuite.addexpectation(

        gx.expectations.ExpectColumnValuesToNotBeNull(column="target")
    )
    expectationsuite.addexpectation(

        gx.expectations.ExpectColumnValuesToBeBetween(
            column="feature1", minvalue=0, maxvalue=100
        )
    )

    validationresult = batch.validate(expectationsuite)
    return validationresult.success

from flytekit import task, ImageSpec

customimage = ImageSpec(
    name="ml-training",
    packages=[
        "torch==2.1.0",
        "transformers==4.36.0",
        "scikit-learn==1.3.0",
    ],
    pythonversion="3.11",

    cuda="12.1",
    registry="ghcr.io/myorg",
)

@task(containerimage=customimage)

def traindeeplearningmodel(
    datasetpath: str,

    epochs: int = 10,
    learningrate: float = 1e-4,
) -> FlyteFile:
    """Training model deep learning di GPU container."""
    import torch
    from transformers import AutoModelForSequenceClassification, Trainer

    model = AutoModelForSequenceClassification.frompretrained(

        "bert-base-uncased", numlabels=2
    )

    modelpath = "/tmp/dlmodel"
    model.savepretrained(modelpath)
    return FlyteFile(path=modelpath)

from flytekit import LaunchPlan, CronSchedule

dailytraining = LaunchPlan.getorcreate(
    name="dailytrainingpipeline",
    workflow=mltrainingpipeline,
    schedule=CronSchedule(schedule="0 2   "),
    defaultinputs={

        "filepath": "/data/dailydataset.csv",

        "config": TrainingConfig(nestimators=200),
    },
)

Notifications

Konfigurasi notifikasi ketika workflow selesai atau gagal:

from flytekit import LaunchPlan, Email

monitoredpipeline = LaunchPlan.getorcreate(

    name="monitoredpipeline",
    workflow=mltrainingpipeline,
    notifications=[
        Email(
            phases=[
                WorkflowExecutionPhase.FAILED,
                WorkflowExecutionPhase.SUCCEEDED,
            ],
            recipientsemail=["team@company.com"],

        )
    ],
)

@task
def goodtask(data: pd.DataFrame, threshold: float) -> list[str]:
    return [col for col in data.columns if data[col].mean() > threshold]

2. Manfaatkan Caching untuk Task yang Mahal

Tambahkan cache=True pada task yang membutuhkan waktu lama dan hasilnya deterministik:

@task(cache=True, cacheversion="1.0")
def expensivecomputation(data: pd.DataFrame) -> pd.DataFrame:
    return data.apply(complextransformation)

Ubah cacheversion ketika logika task berubah agar cache lama tidak digunakan.

3. Definisikan Resource Requirements

Tentukan kebutuhan resource untuk setiap task agar Kubernetes bisa meng-alokasikan resource yang tepat:

@task(
    requests=Resources(cpu="2", mem="4Gi", gpu="1"),
    limits=Resources(cpu="4", mem="8Gi", gpu="1"),
)
def gputrainingtask(data: pd.DataFrame) -> FlyteFile:
    pass

4. Gunakan Dataclass untuk Structured Data

Untuk konfigurasi dan hasil yang kompleks, gunakan dataclass:

from dataclasses import dataclass
from mashumaro.mixins.json import DataClassJSONMixin

@dataclass
class ExperimentResult(DataClassJSONMixin):
    modelname: str

    accuracy: float
    parameters: dict
    trainingdurationseconds: float

if name == "main":
    result = mltrainingpipeline(

        filepath="./testdata.csv",

        config=TrainingConfig(nestimators=10),
    )
    print(f"Metrics: {result}")

6. Gunakan ImageSpec untuk Dependency Management

Definisikan dependencies di level task, bukan di level project:

dataimage = ImageSpec(
    name="data-processing",
    packages=["pandas==2.1.0", "numpy==1.26.0"],
)

mlimage = ImageSpec(
    name="ml-training",
    packages=["scikit-learn==1.3.0", "xgboost==2.0.0"],
)

@task(containerimage=dataimage)
def processdata(raw: pd.DataFrame) -> pd.DataFrame:

    pass

@task(containerimage=mlimage)

def train(data: pd.DataFrame) -> FlyteFile:
    pass

@task(retries=3, timeout=timedelta(hours=1))
def flakyapicall(endpoint: str) -> dict:

    """Task dengan retry otomatis jika gagal."""
    import requests
    response = requests.get(endpoint, timeout=30)
    response.raiseforstatus()

    return response.json()

# Build dan push container image
pyflyte --pkgs myproject package --image ghcr.io/myorg/my-project:latest

Register ke cluster
flytectl register files --project my-project --domain production --archive flyte-package.tgz

Struktur Project yang Direkomendasikan

my-flyte-project/
├── workflows/
│   ├── init.py
│   ├── datapipeline.py

│   ├── trainingpipeline.py
│   └── deploymentpipeline.py

├── tasks/
│   ├── init.py
│   ├── datatasks.py
│   ├── modeltasks.py

│   └── evaluationtasks.py
├── configs/
│   ├── trainingconfig.py

│   └── deploymentconfig.py
├── tests/
│   ├── testdatatasks.py
│   └── testtrainingpipeline.py
├── Dockerfile
├── requirements.txt
└── pyproject.toml

Kesimpulan

Flyte adalah platform workflow orchestration yang dirancang khusus untuk kebutuhan machine learning dan data engineering modern. Dengan fitur type safety, caching, versioning, dan native Kubernetes support, Flyte membantu tim data membangun pipeline yang reproducible dan scalable.

Keunggulan utama Flyte dibandingkan alternatif lain:

• Type safety mengurangi error yang baru terdeteksi saat runtime
• Caching bawaan menghemat waktu dan biaya komputasi
• Versioning otomatis memastikan reproducibility
• Dynamic workflows memberikan fleksibilitas untuk kasus yang kompleks
• Multi-tenancy memudahkan kolaborasi antar tim

Untuk memulai, coba bangun pipeline sederhana dengan dua atau tiga task, jalankan secara lokal, lalu deploy ke cluster Flyte sandbox. Seiring kebutuhan berkembang, manfaatkan fitur lanjutan seperti map tasks, conditional workflows, dan integrasi dengan tools ML lainnya.

Dokumentasi lengkap tersedia di situs resmi Flyte, dan komunitas yang aktif di Slack siap membantu jika Anda menemui kendala.