Kedro: Pipeline Data Science yang Reproducible dan Mudah Dirawat

Sebagian besar proyek data science dimulai dari satu notebook dan perlahan berubah menjadi kekusutan sel, path file yang ditulis langsung (hardcoded), serta kode yang hanya bisa berjalan di satu laptop. Kedro adalah framework Python open-source yang menerapkan disiplin software engineering pada pekerjaan data science, memberi proyek Anda struktur standar, cara deklaratif untuk mengelola data, serta pemisahan yang rapi antara kode, konfigurasi, dan data. Tutorial ini membahas konsep-konsepnya dan menunjukkan contoh lengkap yang dapat dijalankan.

Apa Itu Kedro dan Masalah yang Diselesaikannya

Kedro bukan scheduler dan bukan pula alat notebook. Ia adalah framework proyek: cara terstruktur untuk menata kode data science agar reproducible, dapat diuji, dan siap diserahkan ke engineer lain. Kedro dibuat di QuantumBlack (bagian dari McKinsey) dan kini dikelola di bawah LF AI & Data Foundation.

Siapa pun yang pernah merawat kode data science pasti mengenali masalah-masalah berulang berikut:

• Kekacauan notebook. Logika tersebar di sel-sel yang harus dijalankan dengan urutan tepat. State tersembunyi membuat hasil mustahil direproduksi.
• Path yang hardcoded. pd.readcsv("/Users/ruby/Downloads/datav3final.csv") hanya jalan di satu mesin dan langsung rusak saat rekan kerja meng-clone repository.
• Tanggung jawab yang tercampur. Connection string, hyperparameter model, dan logika bisnis bercampur dalam satu file, sehingga mengganti environment berarti mengubah kode sumber.
• Lineage yang tidak jelas. Tidak jelas fungsi mana menghasilkan artefak mana dan apa bergantung pada apa.

python -m venv .venv
source .venv/bin/activate        # di Windows: .venv\Scripts\activate
pip install kedro
kedro info

kedro new --name price-prediction

price-prediction/
├── conf/
│   ├── base/
│   │   ├── catalog.yml          # definisi dataset
│   │   ├── parameters.yml       # parameter pipeline
│   │   └── logging.yml
│   └── local/
│       └── credentials.yml      # rahasia, jangan pernah di-commit
├── data/
│   ├── 01raw/                  # data sumber yang immutable
│   ├── 02intermediate/         # data yang sudah bertipe dan bersih

│   ├── 03primary/
│   ├── 04feature/

│   ├── 05modelinput/

│   ├── 06models/
│   ├── 07modeloutput/
│   └── 08reporting/

├── notebooks/
├── src/
│   └── priceprediction/
│       ├── pipelines/
│       ├── pipelineregistry.py

│       └── settings.py
├── pyproject.toml
└── requirements.txt

# conf/base/catalog.yml

housesraw:
  type: pandas.CSVDataset
  filepath: data/01raw/houses.csv


housescleaned:
  type: pandas.ParquetDataset
  filepath: data/02intermediate/housescleaned.parquet

modelinputtable:
  type: pandas.ParquetDataset
  filepath: data/05modelinput/modelinput.parquet


pricemodel:
  type: pickle.PickleDataset
  filepath: data/06models/pricemodel.pkl
  versioned: true

Beberapa hal yang perlu disorot:

• Tidak ada path hardcoded di kode. Sebuah node menerima DataFrame yang sudah ter-load dan mengembalikan DataFrame; ia tidak pernah tahu di mana data berada. Untuk berpindah dari CSV ke Parquet, atau dari disk lokal ke S3, Anda mengubah YAML, bukan Python.
• Banyak tipe dataset. Kedro menyediakan konektor untuk CSV, Parquet, Excel, JSON, pickle, tabel dan query SQL, Spark, data citra, dan lainnya, melalui paket kedro-datasets.
• Versioning. Mengatur versioned: true membuat Kedro menulis output setiap run ke subfolder bertimestamp, sehingga artefak model tidak pernah tertimpa diam-diam.
• MemoryDataset. Output apa pun yang tidak dideklarasikan di catalog otomatis disimpan di memori sebagai MemoryDataset dan diteruskan ke node berikutnya dalam run yang sama. Ini praktis untuk hasil antara sementara yang tidak perlu Anda persistkan.

Contoh SQL dan contoh cloud:

# Membaca dari database
customertable:

  type: pandas.SQLTableDataset
  credentials: dbcredentials
  tablename: customers

  loadargs:
    schema: public

Menulis ke cloud storage
predictions:
  type: pandas.CSVDataset
  filepath: s3://my-bucket/predictions/output.csv
  credentials: awscreds

Kredensial yang dirujuk berdasarkan nama (dbcredentials, awscreds) diselesaikan dari conf/local/credentials.yml, menjaga rahasia tetap di luar catalog yang di-commit.

# src/priceprediction/pipelines/dataprocessing/nodes.py
import pandas as pd


def cleanhouses(housesraw: pd.DataFrame) -> pd.DataFrame:

    df = housesraw.copy()
    df = df.dropna(subset=["price", "areasqm"])

    df["pricepersqm"] = df["price"] / df["areasqm"]
    df["hasgarage"] = df["garage"].fillna(0).astype(int)

    return df


def buildmodelinput(housescleaned: pd.DataFrame) -> pd.DataFrame:
    features = ["areasqm", "bedrooms", "bathrooms", "hasgarage", "pricepersqm"]
    return housescleaned[features + ["price"]]

# src/priceprediction/pipelines/dataprocessing/pipeline.py
from kedro.pipeline import Pipeline, node, pipeline
from .nodes import cleanhouses, buildmodelinput


def createpipeline(kwargs) -> Pipeline:

    return pipeline(
        [
            node(
                func=cleanhouses,
                inputs="housesraw",

                outputs="housescleaned",
                name="cleanhousesnode",
            ),
            node(
                func=buildmodelinput,
                inputs="housescleaned",

                outputs="modelinputtable",

                name="buildmodelinputnode",
            ),
        ]
    )

# src/priceprediction/pipelines/datascience/nodes.py
import logging
import pandas as pd
from sklearn.ensemble import RandomForestRegressor
from sklearn.modelselection import traintestsplit

from sklearn.metrics import meanabsoluteerror, r2score

logger = logging.getLogger(name)


def splitdata(modelinputtable: pd.DataFrame, parameters: dict):

    X = modelinputtable.drop(columns=["price"])

    y = modelinputtable["price"]

    return traintestsplit(

        X, y,
        testsize=parameters["testsize"],

        randomstate=parameters["randomstate"],

    )


def trainmodel(Xtrain, ytrain, parameters: dict) -> RandomForestRegressor:
    model = RandomForestRegressor(
        nestimators=parameters["nestimators"],
        maxdepth=parameters["maxdepth"],
        randomstate=parameters["randomstate"],
    )
    model.fit(Xtrain, ytrain)
    return model


def evaluatemodel(model: RandomForestRegressor, Xtest, ytest) -> dict:

    preds = model.predict(Xtest)
    mae = meanabsoluteerror(ytest, preds)

    r2 = r2score(ytest, preds)

    logger.info("MAE model: %.2f, R2: %.3f", mae, r2)
    return {"mae": mae, "r2": r2}

# src/priceprediction/pipelines/datascience/pipeline.py
from kedro.pipeline import Pipeline, node, pipeline
from .nodes import splitdata, trainmodel, evaluatemodel


def createpipeline(kwargs) -> Pipeline:

    return pipeline(
        [
            node(
                func=splitdata,
                inputs=["modelinputtable", "params:modeloptions"],

                outputs=["Xtrain", "Xtest", "ytrain", "ytest"],

                name="splitdatanode",

            ),
            node(
                func=trainmodel,
                inputs=["Xtrain", "ytrain", "params:modeloptions"],

                outputs="pricemodel",
                name="trainmodelnode",
            ),
            node(
                func=evaluatemodel,

                inputs=["pricemodel", "Xtest", "ytest"],
                outputs="modelmetrics",

                name="evaluatemodelnode",

            ),
        ]
    )

Mendaftarkan Pipeline dan Pipeline Default

Kedro menemukan pipeline melalui pipelineregistry.py. Di sini Anda merakit pipeline proyek lengkap, termasuk default yang berjalan ketika tidak ada pipeline spesifik yang disebut.

# src/priceprediction/pipelineregistry.py
from kedro.pipeline import Pipeline
from priceprediction.pipelines import dataprocessing as dp

from priceprediction.pipelines import datascience as ds



def registerpipelines() -> dict[str, Pipeline]:
    dataprocessing = dp.createpipeline()
    datascience = ds.createpipeline()
    return {
        "dp": dataprocessing,

        "ds": datascience,
        "default": dataprocessing + datascience,
    }

Menambahkan dua pipeline dengan + menggabungkannya menjadi satu graph. Karena modelinputtable yang dihasilkan pipeline pertama dikonsumsi oleh pipeline kedua, Kedro secara otomatis mengurutkan keseluruhannya dengan benar dari ujung ke ujung.

Menjalankan Pipeline

Perintah dasar menjalankan pipeline default:

kedro run

Kedro menawarkan kontrol granular atas apa yang dijalankan:

# Jalankan hanya pipeline data science
kedro run --pipeline ds

Jalankan node tertentu berdasarkan nama
kedro run --nodes "trainmodelnode,evaluatemodelnode"

Jalankan dari satu node ke depan, atau sampai sebuah node
kedro run --from-nodes splitdatanode
kedro run --to-nodes buildmodelinputnode


Jalankan semua yang diberi tag untuk keperluan tertentu
kedro run --tags training

# conf/base/parameters.yml
modeloptions:
  testsize: 0.2

  randomstate: 42
  nestimators: 200

  maxdepth: 12

Referensi params:modeloptions pada input sebuah node menyuntikkan dictionary ini ke dalam fungsi. Untuk merujuk satu nilai saja Anda bisa memakai akses bertitik, misalnya params:modeloptions.testsize.

kedro run --env staging

# conf/local/credentials.yml
dbcredentials:
  con: postgresql://user:password@host:5432/prod
awscreds:

  clientkwargs:
    awsaccesskeyid: AKIA...

    awssecretaccesskey: ...

Kedro memakai OmegaConf sebagai config loader bawaannya, yang mendukung interpolasi variabel dan template yang dapat dipakai ulang. Anda bisa mendefinisikan anchor sekali dan memakainya ulang di seluruh catalog:

# conf/base/catalog.yml
parquet: &parquet
  type: pandas.ParquetDataset
  saveargs:
    compression: snappy

housescleaned:

  <<: parquet

  filepath: data/02intermediate/housescleaned.parquet


modelinputtable:

  <<: parquet
  filepath: data/05modelinput/modelinput.parquet

Anda juga dapat menginterpolasi variabel environment dan parameter ke dalam catalog, sehingga konfigurasi tetap ringkas dan konsisten.

Memvisualisasikan Pipeline dengan Kedro-Viz

Kedro-Viz merender DAG Anda sebagai diagram interaktif, menampilkan node, dataset, dan bagaimana data mengalir di antaranya.

pip install kedro-viz
kedro viz run

Perintah ini membuka browser di http://127.0.0.1:4141 dengan graph lengkap. Berguna untuk onboarding anggota tim baru, mendeteksi dependensi yang tidak diinginkan, dan menjelaskan alur kerja ke pihak non-teknis. Kedro-Viz juga dapat menampilkan metrik experiment tracking dari waktu ke waktu ketika Anda mencatatnya melalui catalog.

Bekerja Secara Interaktif: Catalog, IPython, dan Jupyter

Kedro dirancang untuk berdampingan dengan notebook, bukan melarangnya. Beberapa perintah menjembatani kedua dunia:

# Daftar dan inspeksi dataset yang dideklarasikan
kedro catalog list

Luncurkan sesi IPython dengan catalog, context, dan pipeline sudah dimuat
kedro ipython

Luncurkan Jupyter dengan sesi Kedro tersedia
kedro jupyter notebook

Di dalam sesi IPython atau Jupyter Anda mendapat objek catalog siap pakai, sehingga eksplorasi memakai dataset yang sama dengan pipeline:

houses = catalog.load("housesraw")
houses.describe()

Simpan hasil eksperimen kembali melalui catalog
catalog.save("housescleaned", cleaneddf)

# src/priceprediction/hooks.py
import logging
from kedro.framework.hooks import hookimpl


logger = logging.getLogger(name)


class ModelTrackingHooks:
    @hookimpl
    def afternoderun(self, node, outputs):
        if node.name == "evaluatemodelnode":
            metrics = outputs.get("modelmetrics", {})

            logger.info("Mencatat metrik ke sistem tracking: %s", metrics)

# src/priceprediction/settings.py
from priceprediction.hooks import ModelTrackingHooks


HOOKS = (ModelTrackingHooks(),)

kedro package

  pip install kedro-airflow

  kedro airflow create
  

• Jaga fungsi node tetap murni. Tanpa I/O file, tanpa state global, tanpa efek samping tersembunyi. Catalog menangani persistensi; node menangani logika.
• Beri nama pada segalanya. Beri setiap node sebuah name agar dapat dialamatkan dari CLI dan terbaca di Kedro-Viz.
• Satu tanggung jawab per pipeline. Pisahkan pemrosesan data dari data science, dan jaga pipeline tetap kecil serta dapat dirangkai.
• Jangan pernah commit rahasia. Kredensial berada di conf/local, yang tetap di luar version control.
• Manfaatkan lapisan data. Mengikuti konvensi 01raw hingga 08_reporting membuat lineage mendokumentasikan dirinya sendiri.
• Versikan model Anda. Atur versioned: true pada dataset model dan output agar run dapat diaudit.
• Uji node secara langsung. Karena fungsi murni, unit test tidak memerlukan context Kedro.
• Promosikan kode notebook menjadi node. Bereksplorasilah di notebook, tetapi pindahkan apa pun yang penting ke dalam pipeline.

Kesimpulan dan Poin Utama

Kedro membawa struktur software engineering ke data science tanpa memaksa Anda meninggalkan gaya eksploratif yang membuat bidang ini produktif. Dengan memisahkan kode, konfigurasi, dan data, serta merangkai fungsi-fungsi murni menjadi pipeline deklaratif, Kedro mengubah notebook sekali pakai menjadi proyek yang bisa di-clone, dijalankan, dan dipercaya oleh rekan kerja mana pun.

Poin utama:

• Kedro adalah framework proyek, bukan scheduler. Ia mendefinisikan struktur; alat seperti Airflow, Dagster, dan Argo menyediakan penjadwalan.
• Data Catalog menghapus path hardcoded dan memusatkan setiap definisi dataset dalam YAML.
• Node adalah fungsi murni; pipeline merangkainya menjadi graph yang urutannya disimpulkan Kedro secara otomatis.
• Konfigurasi, parameter, dan kredensial dipisah per environment, tidak pernah di-hardcode.
• Kedro-Viz memvisualisasikan DAG, dan hooks/plugin memperluas perilaku serta berintegrasi dengan MLflow dan target deployment.
• kedro package menghasilkan artefak portabel yang bisa Anda jalankan di orkestrator pilihan Anda.

Mulailah dari yang kecil: modelkan satu pipeline dengan dua atau tiga node, deklarasikan dataset-nya di catalog, dan jalankan dengan kedro run. Disiplin ini langsung berbuah begitu orang kedua perlu menyentuh proyek tersebut.