Tutorial Metaflow: Framework MLOps dari Netflix untuk Data Science

Metaflow adalah framework open-source yang dikembangkan oleh Netflix untuk membangun dan mengelola proyek data science secara efisien. Framework ini memungkinkan data scientist untuk fokus pada pemodelan tanpa harus khawatir tentang infrastruktur, orkestrasi, dan deployment. Dalam tutorial ini, kita akan mempelajari cara menggunakan Metaflow dari dasar hingga fitur-fitur lanjutan.

Mengapa Metaflow?

Dalam dunia data science dan machine learning, salah satu tantangan terbesar bukan membangun model, melainkan mengelola seluruh lifecycle dari eksperimen hingga produksi. Metaflow hadir untuk menyelesaikan masalah ini dengan pendekatan yang pragmatis:

• Human-centric API: Dirancang agar mudah digunakan oleh data scientist, bukan hanya ML engineers
• Versioning otomatis: Setiap eksperimen secara otomatis di-track dan bisa direproduksi
• Skalabilitas transparan: Beralih dari laptop ke cloud tanpa mengubah kode
• Dependency management: Mengelola environment Python secara otomatis
• Integration: Bekerja dengan AWS, Azure, GCP, dan Kubernetes

Instalasi

Instalasi Dasar

pip install metaflow

Instalasi dengan Dukungan AWS

pip install metaflow[aws]

Verifikasi Instalasi

import metaflow
print(metaflow.version)

Konfigurasi Awal

Setelah instalasi, jalankan konfigurasi:

metaflow configure show

Untuk konfigurasi AWS S3 sebagai datastore:

metaflow configure aws

Konsep Dasar: Flow dan Step

Metaflow menggunakan konsep Flow dan Step untuk mengorganisasi pipeline data science. Sebuah Flow adalah DAG (Directed Acyclic Graph) yang terdiri dari beberapa Step.

Flow Pertama

from metaflow import FlowSpec, step

class HelloFlow(FlowSpec):

    @step
    def start(self):
        print("Memulai flow pertama!")
        self.message = "Hello from Metaflow"
        self.next(self.end)

    @step
    def end(self):
        print(f"Pesan: {self.message}")
        print("Flow selesai!")

if name == 'main':
    HelloFlow()

Jalankan flow:

python helloflow.py run

from metaflow import FlowSpec, step

class ArtifactFlow(FlowSpec):

    @step
    def start(self):
        self.data = [1, 2, 3, 4, 5]
        self.modelname = "randomforest"

        self.next(self.process)

    @step
    def process(self):
        self.result = sum(self.data)  2

        print(f"Model: {self.modelname}")
        print(f"Hasil: {self.result}")
        self.next(self.end)

    @step
    def end(self):
        print(f"Final result: {self.result}")

if name == 'main':
    ArtifactFlow()

from metaflow import FlowSpec, step

class BranchFlow(FlowSpec):

    @step
    def start(self):
        self.rawdata = list(range(100))

        self.next(self.trainrf, self.trainxgb)


    @step
    def trainrf(self):
        from sklearn.ensemble import RandomForestClassifier
        from sklearn.datasets import makeclassification


        X, y = makeclassification(nsamples=1000, nfeatures=20, randomstate=42)

        model = RandomForestClassifier(nestimators=100, randomstate=42)

        model.fit(X, y)
        self.accuracy = model.score(X, y)
        self.modeltype = "RandomForest"
        print(f"RF Accuracy: {self.accuracy:.4f}")
        self.next(self.join)

    @step
    def trainxgb(self):

        from sklearn.ensemble import GradientBoostingClassifier
        from sklearn.datasets import makeclassification

        X, y = makeclassification(nsamples=1000, nfeatures=20, randomstate=42)
        model = GradientBoostingClassifier(nestimators=100, randomstate=42)
        model.fit(X, y)
        self.accuracy = model.score(X, y)
        self.modeltype = "GradientBoosting"

        print(f"XGB Accuracy: {self.accuracy:.4f}")
        self.next(self.join)

    @step
    def join(self, inputs):
        best = max(inputs, key=lambda x: x.accuracy)
        self.bestmodel = best.modeltype

        self.bestaccuracy = best.accuracy
        print(f"Model terbaik: {self.bestmodel} ({self.bestaccuracy:.4f})")
        self.next(self.end)

    @step
    def end(self):
        print(f"Pipeline selesai. Model terpilih: {self.bestmodel}")


if name == 'main':
    BranchFlow()

from metaflow import FlowSpec, step, Parameter

class ParamFlow(FlowSpec):

    learningrate = Parameter(
        'learningrate',

        help='Learning rate untuk training',
        default=0.01
    )

    nestimators = Parameter(
        'nestimators',

        help='Jumlah estimator',
        default=100,
        type=int
    )

    datapath = Parameter(
        'datapath',

        help='Path ke dataset',
        default='data/train.csv'
    )

    @step
    def start(self):
        print(f"Learning Rate: {self.learningrate}")
        print(f"N Estimators: {self.nestimators}")

        print(f"Data Path: {self.datapath}")
        self.next(self.train)

    @step
    def train(self):
        from sklearn.ensemble import GradientBoostingClassifier
        from sklearn.datasets import makeclassification


        X, y = makeclassification(nsamples=1000, randomstate=42)
        model = GradientBoostingClassifier(
            learningrate=self.learningrate,
            nestimators=self.nestimators,
            randomstate=42

        )
        model.fit(X, y)
        self.score = model.score(X, y)
        print(f"Score: {self.score:.4f}")
        self.next(self.end)

    @step
    def end(self):
        print(f"Training selesai dengan score: {self.score:.4f}")

if name == 'main':
    ParamFlow()

python paramflow.py run --learningrate 0.1 --nestimators 200

from metaflow import FlowSpec, step

class ForeachFlow(FlowSpec):

    @step
    def start(self):
        self.datasets = ['dataseta', 'datasetb', 'datasetc']

        self.next(self.process, foreach='datasets')

    @step
    def process(self):
        datasetname = self.input
        print(f"Memproses: {datasetname}")


        import random
        random.seed(hash(datasetname) % 232)

        self.accuracy = random.uniform(0.7, 0.99)
        self.dataset = datasetname

        print(f"{datasetname}: accuracy = {self.accuracy:.4f}")
        self.next(self.join)

    @step
    def join(self, inputs):
        self.results = {
            inp.dataset: inp.accuracy for inp in inputs
        }
        best = max(inputs, key=lambda x: x.accuracy)
        self.bestdataset = best.dataset

        self.bestaccuracy = best.accuracy

        print("\nHasil semua dataset:")
        for name, acc in self.results.items():
            print(f"  {name}: {acc:.4f}")
        print(f"\nTerbaik: {self.bestdataset} ({self.bestaccuracy:.4f})")
        self.next(self.end)

    @step
    def end(self):
        print("Pipeline foreach selesai!")

if name == 'main':
    ForeachFlow()

from metaflow import FlowSpec, step, condabase, pypi

@condabase(python='3.10')
class MLFlow(FlowSpec):

    @pypi(packages={'scikit-learn': '1.4.0', 'pandas': '2.2.0'})
    @step
    def start(self):
        import sklearn
        import pandas as pd
        print(f"sklearn: {sklearn.version}")
        print(f"pandas: {pd.version}")
        self.next(self.end)

    @step
    def end(self):
        print("Selesai!")

if name == 'main':
    MLFlow()

from metaflow import FlowSpec, step, resources

class GPUFlow(FlowSpec):

    @resources(cpu=4, memory=8000)
    @step
    def start(self):
        self.data = list(range(1000000))
        self.next(self.train)

    @resources(cpu=8, memory=16000, gpu=1)
    @step
    def train(self):
        print("Training dengan GPU...")
        self.modeltrained = True

        self.next(self.end)

    @step
    def end(self):
        print(f"Model trained: {self.modeltrained}")

if name == 'main':
    GPUFlow()

from metaflow import FlowSpec, step, retry, catch

class RobustFlow(FlowSpec):

    @retry(times=3)
    @step
    def start(self):
        self.data = self.loaddata()

        self.next(self.train)

    def loaddata(self):
        import random
        if random.random() < 0.3:
            raise Exception("Koneksi terputus!")
        return [1, 2, 3, 4, 5]

    @catch(var='trainerror')

    @step
    def train(self):
        self.result = sum(self.data)  2
        self.next(self.end)

    @step
    def end(self):
        if hasattr(self, 'trainerror') and self.trainerror:

            print(f"Training gagal: {self.trainerror}")
        else:
            print(f"Hasil: {self.result}")

if name == 'main':
    RobustFlow()

Client API: Mengakses Hasil Eksperimen

Salah satu fitur paling powerful dari Metaflow adalah Client API untuk mengakses hasil eksperimen sebelumnya:

from metaflow import Flow, Run, Step

flow = Flow('BranchFlow')

for run in flow.runs():
    print(f"Run ID: {run.id}")
    print(f"  Status: {run.successful}")
    print(f"  Waktu: {run.createdat}")


    if run.successful:
        endstep = Step(f'BranchFlow/{run.id}/end')
        for task in endstep.tasks():

            print(f"  Best Model: {task['bestmodel'].data}")
            print(f"  Best Accuracy: {task['bestaccuracy'].data}")

    print()

from metaflow import Flow

run = Flow('BranchFlow').latestsuccessfulrun

print(f"Run terakhir: {run.id}")
print(f"Best model: {run.data.bestmodel}")
print(f"Best accuracy: {run.data.bestaccuracy}")

from metaflow import FlowSpec, step, Parameter, retry, catch
import json

class MLPipelineFlow(FlowSpec):

    testsize = Parameter('testsize', default=0.2, type=float)

    randomstate = Parameter('randomstate', default=42, type=int)


    @step
    def start(self):
        print("=== ML Pipeline Dimulai ===")
        self.next(self.loaddata)

    @retry(times=2)
    @step
    def loaddata(self):

        from sklearn.datasets import loadbreastcancer

        import pandas as pd

        data = loadbreastcancer()

        self.df = pd.DataFrame(data.data, columns=data.featurenames)
        self.target = list(data.target)
        self.featurenames = list(data.featurenames)

        print(f"Dataset loaded: {self.df.shape}")
        print(f"Features: {len(self.featurenames)}")

        self.next(self.preprocess)

    @step
    def preprocess(self):
        from sklearn.modelselection import traintestsplit
        from sklearn.preprocessing import StandardScaler
        import numpy as np

        X = self.df.values
        y = np.array(self.target)

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

        scaler = StandardScaler()
        self.Xtrain = scaler.fittransform(Xtrain).tolist()

        self.Xtest = scaler.transform(Xtest).tolist()

        self.ytrain = ytrain.tolist()

        self.ytest = ytest.tolist()

        self.scalermean = scaler.mean.tolist()

        self.scalerscale = scaler.scale.tolist()


        print(f"Train: {len(self.Xtrain)}, Test: {len(self.Xtest)}")

        self.next(self.trainlr, self.trainrf, self.trainsvm)

    @step
    def trainlr(self):

        from sklearn.linearmodel import LogisticRegression
        from sklearn.metrics import accuracyscore, f1score
        import numpy as np

        Xtrain = np.array(self.Xtrain)
        ytrain = np.array(self.ytrain)
        Xtest = np.array(self.Xtest)
        ytest = np.array(self.ytest)

        model = LogisticRegression(maxiter=1000, randomstate=self.randomstate)

        model.fit(Xtrain, ytrain)


        ypred = model.predict(Xtest)

        self.accuracy = accuracyscore(ytest, ypred)
        self.f1 = f1score(ytest, ypred)

        self.modelname = "LogisticRegression"
        print(f"LR - Accuracy: {self.accuracy:.4f}, F1: {self.f1:.4f}")
        self.next(self.compare)

    @step
    def trainrf(self):

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

        import numpy as np

        Xtrain = np.array(self.Xtrain)

        ytrain = np.array(self.ytrain)

        Xtest = np.array(self.Xtest)

        ytest = np.array(self.ytest)


        model = RandomForestClassifier(
            nestimators=100, randomstate=self.randomstate
        )
        model.fit(Xtrain, ytrain)

        ypred = model.predict(Xtest)
        self.accuracy = accuracyscore(ytest, ypred)

        self.f1 = f1score(ytest, ypred)
        self.modelname = "RandomForest"

        print(f"RF - Accuracy: {self.accuracy:.4f}, F1: {self.f1:.4f}")
        self.next(self.compare)

    @step
    def trainsvm(self):
        from sklearn.svm import SVC
        from sklearn.metrics import accuracyscore, f1score
        import numpy as np

        Xtrain = np.array(self.Xtrain)
        ytrain = np.array(self.ytrain)
        Xtest = np.array(self.Xtest)
        ytest = np.array(self.ytest)

        model = SVC(kernel='rbf', randomstate=self.randomstate)
        model.fit(Xtrain, ytrain)

        ypred = model.predict(Xtest)
        self.accuracy = accuracyscore(ytest, ypred)

        self.f1 = f1score(ytest, ypred)
        self.modelname = "SVM"

        print(f"SVM - Accuracy: {self.accuracy:.4f}, F1: {self.f1:.4f}")
        self.next(self.compare)

    @step
    def compare(self, inputs):
        results = []
        for inp in inputs:
            results.append({
                'model': inp.modelname,
                'accuracy': inp.accuracy,
                'f1': inp.f1
            })

        results.sort(key=lambda x: x['f1'], reverse=True)

        print("\n=== Perbandingan Model ===")
        for r in results:
            print(f"  {r['model']}: Accuracy={r['accuracy']:.4f}, F1={r['f1']:.4f}")

        best = results[0]
        self.bestmodelname = best['model']
        self.bestaccuracy = best['accuracy']

        self.bestf1 = best['f1']
        self.allresults = results


        print(f"\nModel Terbaik: {self.bestmodelname}")

        self.next(self.end)

    @step
    def end(self):
        print("\n=== ML Pipeline Selesai ===")
        print(f"Model terpilih: {self.bestmodelname}")

        print(f"Accuracy: {self.bestaccuracy:.4f}")
        print(f"F1 Score: {self.bestf1:.4f}")


if name == 'main':
    MLPipelineFlow()

python mlpipeline.py run

Dengan parameter kustom:

python mlpipeline.py run --testsize 0.3 --randomstate 123

python mlpipeline.py argo-workflows create

python mlpipeline.py argo-workflows trigger

from metaflow import FlowSpec, step, card, current
from metaflow.cards import Markdown, Table, Image

class CardFlow(FlowSpec):

    @card
    @step
    def start(self):
        self.metrics = {
            'accuracy': 0.95,
            'precision': 0.93,
            'recall': 0.97
        }
        current.card.append(Markdown("# Hasil Training"))
        current.card.append(
            Table([
                ['Metrik', 'Nilai'],
                ['Accuracy', '0.95'],
                ['Precision', '0.93'],
                ['Recall', '0.97']
            ])
        )
        self.next(self.end)

    @step
    def end(self):
        print("Selesai!")

if name == 'main':
    CardFlow()

python cardflow.py card view start

Best Practices

1. Desain Step yang Modular

Setiap step sebaiknya melakukan satu tugas spesifik. Ini memudahkan debugging dan memungkinkan retry per step.

@step
def loaddata(self):

    ...
    self.next(self.validatedata)

@step
def validatedata(self):

    ...
    self.next(self.preprocess)

@step
def train(self):
    self.metrics = {'accuracy': 0.95}
    self.next(self.end)

@retry(times=3)
@step
def fetchdata(self):
    ...

6. Gunakan Namespace untuk Organisasi

METAFLOWUSER=production python flow.py run

METAFLOWUSER=experiment python flow.py run

7. Monitoring dengan Client API

Buat script monitoring untuk memantau status pipeline:

from metaflow import Flow

flow = Flow('MLPipelineFlow')
for run in list(flow.runs())[:5]:
    status = "Berhasil" if run.successful else "Gagal"
    print(f"Run {run.id}: {status} ({run.created_at})")

Kesimpulan

Metaflow adalah framework MLOps yang dirancang untuk menyederhanakan workflow data science dari eksperimen hingga produksi. Dengan fitur-fitur seperti versioning otomatis, branching paralel, dependency management, dan Client API, Metaflow memungkinkan data scientist untuk fokus pada pemodelan tanpa terbebani oleh kompleksitas infrastruktur.

Keunggulan utama Metaflow:

• Kemudahan penggunaan: API yang intuitif dan Pythonic
• Reprodusibilitas: Setiap eksperimen otomatis di-track
• Skalabilitas: Transisi mulus dari laptop ke cloud
• Fleksibilitas: Mendukung berbagai cloud provider dan orchestrator
• Monitoring: Client API dan Cards untuk observability

Untuk langkah selanjutnya, Anda bisa mengeksplorasi integrasi Metaflow dengan AWS Step Functions, Kubernetes, atau Apache Airflow untuk production deployment yang lebih kompleks. Dokumentasi resmi di docs.metaflow.org menyediakan panduan lengkap untuk setiap fitur.