Orkestrasi Alur Kerja ML dengan Prefect

Dalam tutorial ini, kita akan menyelami Prefect dengan mengeksplorasi fitur utama, manfaat, dan komponennya. Kita akan melihat cara membuat task dan flow, mengelola deployment, serta mengeksekusi flow run. Terakhir, kami akan memandu Anda menerapkan alur kerja di Prefect Cloud dan menjalankannya menggunakan infrastruktur cloud.

Bagi Anda yang tertarik mengotomatiskan pipeline machine learning menggunakan GitHub Actions dan Data Version Control, pertimbangkan untuk mengikuti kursus CI/CD for Machine Learning.

Apa itu Prefect?

Prefect adalah alat orkestrasi alur kerja open-source yang andal, memungkinkan pengguna merancang, memantau, dan merespons pipeline data dan machine learning menggunakan kode Python.

Prefect dilengkapi beragam fitur, seperti percobaan ulang otomatis, penjadwalan, dan caching, untuk menyebut beberapa. Ini adalah alat bertenaga yang pada akhirnya memungkinkan Anda membangun alur kerja yang tangguh dan dinamis.

Gambar Sumber

Dengan Prefect, Anda dapat dengan mudah mengubah fungsi Python apa pun menjadi unit kerja yang dapat diamati dan diorkestrasi hanya dengan menambahkan beberapa decorator pada kode Anda.

Prefect ideal untuk standarisasi pengembangan dan deployment alur kerja di seluruh organisasi. Ini memberikan observabilitas menyeluruh pada alur kerja dan memungkinkan pengguna mengelola kode mereka tanpa perlu DAG yang ketat atau boilerplate.

Fitur Utama Prefect

Ringan

Dengan satu perintah, pengguna dapat menyiapkan server orkestrasi mereka sendiri untuk kebutuhan pengembangan. Proses yang efisien ini menyederhanakan pengembangan lokal dan memastikan deployment ke lingkungan produksi berjalan mulus.

UI dan Dasbor yang Intuitif

Prefect menyediakan antarmuka pengguna yang umumnya intuitif, meningkatkan pengalaman pengguna dengan cara yang menarik secara visual dan mudah digunakan untuk menjalankan, memantau, debug, dan mengelola alur kerja.

Fitur Lengkap

Prefect menawarkan fitur seperti penjadwalan, percobaan ulang, manajemen secret native, kemampuan penskalaan, kontrol konkurensi, opsi konfigurasi infrastruktur dan penyimpanan yang fleksibel, serta fungsionalitas logging yang andal.

Fleksibilitas Open-source

Pengguna bebas untuk melakukan self-host server Prefect open-source atau memanfaatkan fitur lanjutan, ketersediaan tinggi, dan keamanan yang ditingkatkan dengan menggunakan Prefect Cloud.

Prefect Cloud

Prefect Cloud menyediakan semua kapabilitas server Prefect open-source beserta fitur tambahan seperti automations, event feed, events, webhook, workspace, organisasi, dan lainnya, yang disesuaikan untuk lingkungan terkelola. Anda dapat menggunakan Prefect Cloud secara gratis atau membayar untuk fitur yang lebih tangguh.

Murni Python

Prefect memungkinkan pengguna memulai orkestrasi alur kerja dengan satu decorator, merampingkan proses tanpa abstraksi yang tidak perlu untuk manajemen alur kerja yang efisien.

Notifikasi

Kirim informasi tentang alur kerja dan pipeline Prefect ke berbagai layanan pesan seperti Slack, Discord, email, Microsoft Teams, dan lainnya. Ini juga dapat mengingatkan Anda jika alur kerja gagal.

Komponen Inti Prefect

Bagian ini membahas komponen utama yang digunakan untuk membangun task dan flow, menerapkan dan mengeksekusinya, serta mengelolanya. Penting untuk meninjau komponen ini sebelum masuk ke tutorial Python.

Prefect Task

Prefect Task adalah fungsi Python yang diberi decorator @task yang merepresentasikan unit kerja diskret dalam alur kerja Prefect. Kita juga dapat menyesuaikan decorator task dengan argumen opsional seperti name, description, tag, pengaturan cache, percobaan ulang, dan lainnya.

Pada kode di bawah, ada tiga task machine learning yang dibuat dengan decorator @task yang dapat digunakan kembali di berbagai flow.

@task
def train_model(X_train, X_test, y_train):
    # Selecting the best features
    ...

    # Train the model
    ...

    return model


@task
def get_prediction(X_test, model: LogisticRegression):
    ...
	return prediction

@task
def evaluate_model(y_test, prediction: pd.DataFrame):
    ...

Prefect Flow

Prefect Flow adalah fungsi Python yang diberi decorator @flow yang merangkum logika alur kerja. Ini memudahkan pengguna untuk mendefinisikan, mengonfigurasi, dan mengeksekusi pipeline data yang kompleks dengan fleksibilitas dan kemudahan.

Pada kode di bawah, kita membuat sebuah Flow menggunakan decorator @flow. Flow akan menjalankan semua task secara berurutan, mengambil input dari satu task ke task berikutnya.

@flow
def ml_workflow():
    model = train_model(X_train, X_test, y_train)
    predictions = get_prediction(X_test, model)
    evaluate_model(y_test, predictions)

if __name__ == "__main__":
    ml_workflow()

Prefect Deployments

Deployment adalah flow yang disimpan di server lokal atau di cloud dan mencakup informasi penting untuk mengorkestrasi alur kerja Anda dari jarak jauh, seperti detail penjadwalan dan eksekusi.

Prefect Work Pools

Work pool berperan sebagai mediator antara lingkungan orkestrasi dan eksekusi, memungkinkan penjadwalan dan eksekusi flow run yang efisien. Pengguna dapat menyiapkan Agent (worker) untuk menjalankan alur kerja secara lokal atau memanfaatkan infrastruktur cloud untuk eksekusi alur kerja yang lancar.

Mirip dengan instance komputasi khusus, Work Pool secara efisien mengelola alokasi pekerjaan dan memprioritaskan task, menawarkan fleksibilitas dalam memilih lingkungan eksekusi untuk kinerja dan skalabilitas alur kerja yang optimal.

Mengorkestrasi Alur Kerja ML

Di bagian ini, kita akan membuat, menguji, menerapkan, dan mengeksekusi alur kerja machine learning sederhana menggunakan Prefect. Kita juga akan menyiapkan notifikasi agar mendapatkan pemberitahuan di server Discord tentang status flow run.

Menyiapkan Prefect

Instal paket Python menggunakan PIP

pip install -U prefect

Jika Anda ingin menggunakan versi Prefect terbaru, Anda bisa mendapatkan kodenya langsung dari GitHub.

pip install -U git+https://github.com/PrefectHQ/prefect

Membuat Prefect Flow

Kita akan melakukan praproses dataset Bank Churn dari Kaggle, lalu menggunakan scikit-learn untuk melatih dan mengevaluasi model Random Forest Classifiers.

Kode yang akan kita gunakan merupakan versi modifikasi dari kode pada: Streamline Your Machine Learning Workflow with Scikit-learn Pipelines. Lihat tutorial tersebut jika Anda tertarik mempelajari lebih lanjut tentang cara model dilatih. Prefect Flow machine learning kita terdiri dari tujuh task:

1. load_data: untuk memuat dan memproses file CSV menggunakan pandas
2. preprocessing: menyiapkan data untuk pelatihan, mengisi nilai yang hilang, melakukan encoding, dan penskalaan.
3. data_split: membagi data menjadi set pengujian dan pelatihan
4. train_model: memilih fitur teratas dan melatih model
5. get_prediction: menghasilkan prediksi menggunakan set pengujian
6. evaluate_model: menghitung akurasi dan skor f1
7. save_model: menyimpan bobot model menggunakan skops

Kini kita telah merangkai semua task yang diperlukan ke dalam fungsi @flow bernama ml_workflow. Selain itu, kami menambahkan argumen tambahan pada decorator untuk mengaktifkan logging terperinci.

Fungsi ml_workflow menerima lokasi file data sebagai input lalu menghubungkan berbagai task secara berantai. Kita akan dapat memvisualisasikan alur kerja ini dengan lebih baik saat menerapkannya dan menjalankannya di server Prefect.

Anda juga dapat melacak eksperimen machine learning kami menggunakan MLflow. Ikuti kursus Introduction to MLflow untuk mempelajari pelacakan MLflow, proyek, model, dan model registry.

import pandas as pd
import skops.io as sio
from prefect import flow, task
from sklearn.feature_selection import SelectKBest, chi2
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, f1_score
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import MinMaxScaler, OrdinalEncoder


@task
def load_data(filename: str):
    bank_df = pd.read_csv(filename, index_col="id", nrows=1000)
    bank_df = bank_df.drop(["CustomerId", "Surname"], axis=1)
    bank_df = bank_df.sample(frac=1)
    return bank_df


@task
def preprocessing(bank_df: pd.DataFrame):
    cat_col = [1, 2]
    num_col = [0, 3, 4, 5, 6, 7, 8, 9]

    # Filling missing categorical values
    cat_impute = SimpleImputer(strategy="most_frequent")
    bank_df.iloc[:, cat_col] = cat_impute.fit_transform(bank_df.iloc[:, cat_col])

    # Filling missing numerical values
    num_impute = SimpleImputer(strategy="median")
    bank_df.iloc[:, num_col] = num_impute.fit_transform(bank_df.iloc[:, num_col])

    # Encode categorical features as an integer array.
    cat_encode = OrdinalEncoder()
    bank_df.iloc[:, cat_col] = cat_encode.fit_transform(bank_df.iloc[:, cat_col])

    # Scaling numerical values.
    scaler = MinMaxScaler()
    bank_df.iloc[:, num_col] = scaler.fit_transform(bank_df.iloc[:, num_col])
    return bank_df


@task
def data_split(bank_df: pd.DataFrame):
    # Splitting data into training and testing sets
    X = bank_df.drop(["Exited"], axis=1)
    y = bank_df.Exited

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.3, random_state=125
    )
    return X_train, X_test, y_train, y_test
    # Identify numerical and categorical columns


@task
def train_model(X_train, X_test, y_train):
    # Selecting the best features
    KBest = SelectKBest(chi2, k="all")
    X_train = KBest.fit_transform(X_train, y_train)
    X_test = KBest.transform(X_test)

    # Train the model
    model = LogisticRegression(max_iter=1000, random_state=125)
    model.fit(X_train, y_train)

    return model


@task
def get_prediction(X_test, model: LogisticRegression):
    return model.predict(X_test)


@task
def evaluate_model(y_test, prediction: pd.DataFrame):
    accuracy = accuracy_score(y_test, prediction)
    f1 = f1_score(y_test, prediction, average="macro")

    print("Accuracy:", str(round(accuracy, 2) * 100) + "%", "F1:", round(f1, 2))


@task
def save_model(model: LogisticRegression):
    sio.dump(model, "bank_model.skops")


@flow(log_prints=True)
def ml_workflow(filename: str = "train.csv"):
    data = load_data(filename)
    prep_data = preprocessing(data)
    X_train, X_test, y_train, y_test = data_split(prep_data)
    model = train_model(X_train, X_test, y_train)
    predictions = get_prediction(X_test, model)
    evaluate_model(y_test, predictions)
    save_model(model)


if __name__ == "__main__":
    ml_workflow()

Untuk menguji Prefect Flow kita, jalankan file Python di terminal.

$ python main.py

Flow berhasil diselesaikan. Ini telah menampilkan metrik model dan menyimpan bobot model.

Jika Anda baru dalam machine learning dan ingin mempelajari cara melakukan praproses data, melatih model, mengoptimalkannya, mengevaluasi, dan menyimpan model, pertimbangkan untuk mengikuti jalur karier Machine Learning Scientist with Python. Kursus ini akan mengajarkan semua yang perlu Anda ketahui dari awal.

Menerapkan Flow

Membangun dan menjalankan pipeline machine learning secara manual adalah tugas sederhana yang bahkan dapat dilakukan dengan pipeline scikit-learn.

Namun, ada banyak manfaat menggunakan Prefect. Ini termasuk:

• kemampuan memantau pipeline,
• menjadwalkannya menggunakan CRON,
• secara otomatis mencoba ulang jika terjadi kegagalan,
• mengaktifkan logging dan observabilitas,
• menerima notifikasi,
• serta membuat alur kerja otomatis yang dapat berjalan tanpa intervensi manusia.

Sekarang, kita akan belajar mengotomatiskan alur kerja dengan menerapkannya ke server Prefect lokal.

Pertama, kita harus membangun “Deployment” dengan memberikan nama file dan nama fungsi flow. Kita juga menambahkan tag “dev” pada Deployment kita.

Catatan: tag di Prefect dapat digunakan untuk mengategorikan dan mengatur flow, task, dan deployment dalam ekosistem Prefect.

$ prefect deployment build main.py:ml_workflow -n 'ml_workflow_bank_churn' -a --tag dev

Deployment berhasil dibuat. Sekarang, diberikan instruksi tentang cara mengeksekusi flow menggunakan Prefect Agent.

Found flow 'ml-workflow'
Deployment YAML created at 
'C:\Repository\GitHub\ML-Workflow-Orchestration-With-Prefect\ml_workflow-deployment.yaml'.
Deployment storage None does not have upload capabilities; no files uploaded.  Pass --skip-upload to 
suppress this warning.
Deployment 'ml-workflow/ml_workflow_bank_churn' successfully created with id 
'ce30d1c7-f454-4064-870c-429a8039c194'.

To execute flow runs from this deployment, start an agent that pulls work from the
'default-agent-pool' work pool:
$ prefect agent start -p 'default-agent-pool'

Menjalankan Deployment Secara Lokal

Prefect agent bertanggung jawab memantau work pool untuk mengidentifikasi flow run yang tersedia dan menerapkannya ke lingkungan eksekusi.

Satu work pool dapat memiliki banyak agent, dengan setiap agent memiliki ID unik untuk mencegah tumpang tindih dan memastikan penskalaan berjalan mulus.

Sekarang kita akan menginisialisasi Prefect agent di terminal baru dengan nama work pool default.

$ prefect agent start -p 'default-agent-pool'

Buka terminal baru dan jalankan deployment dengan memberikan “nama fungsi flow/nama deployment.” Anda juga bisa hanya menggunakan nama deployment untuk mengeksekusi flow.

$ prefect deployment run 'ml-workflow/ml_workflow_bank_churn'

Ini telah membuat flow run untuk deployment yang akan memeriksa worker dan mulai menjalankan alur kerja.

Creating flow run for deployment 'ml-workflow/ml_workflow_bank_churn'...
Created flow run 'cryptic-potoo'.
└── UUID: 77833ee3-0f7e-42f9-8002-0a562c7d3ed4
└── Parameters: {}
└── Scheduled start time: 2024-03-14 21:30:40 PKT (now)
└── URL: <no dashboard available>

Beralihlah ke terminal tempat kita memulai agent untuk melihat log flow run kita. Prefect agent telah berhasil mengeksekusi semua task dalam flow.

Ini adalah contoh sederhana. Kita juga dapat menjadwalkan atau mengotomatiskan alur kerja dengan memberikan argumen tambahan pada build Deployment kita.

Prefect UI

Sejauh ini, kita mengamati alur kerja di terminal. Sekarang, mari luncurkan dasbor Prefect tempat kita dapat melihat semua flow run, deployment, work pool, dan notifikasi.

Mulai server Prefect di terminal baru dan klik tautan lokal yang disediakan untuk mengakses dasbor.

$ prefect server start

Untuk melihat semua flow yang telah berjalan sebelumnya, klik opsi "Flow" di panel kiri. Anda mungkin melihat beberapa flow tidak dieksekusi karena tidak ada work pool, sementara yang lain dieksekusi tanpa tag "dev" selama fase eksperimen.

Mengklik opsi Flows menampilkan semua flow yang tersedia. Beberapa flow dapat ada dalam satu deployment.

Di bagian Deployment, kita dapat melihat Deployment saat ini beserta aktivitas dan tag-nya. Di bagian ini, kita dapat dengan mudah menjalankan flow menggunakan opsi quick run atau menjadwalkan run menggunakan antarmuka pengguna.

Di Work Pool, Anda akan melihat Prefect Agent default kita. Anda bahkan dapat menyiapkan work pool cloud juga dengan menggunakan Prefect CLI.

Menambahkan Notifikasi Discord

Mari mulai dari bagian yang menarik.

Tidak ada yang ingin memantau alur kerja sepanjang waktu. Sebaliknya, kita ingin menerima notifikasi tentang status alur kerja di aplikasi pesan pilihan. Dengan begitu, kita dapat segera menyelesaikan masalah apa pun.

Untuk mencapainya, kita akan membuat webhook Discord. Menggunakan ID dan token webhook, kita akan menyiapkan notifikasi Prefect. Ini akan mengirim semua informasi log ke server Discord.

Untuk membuat webhook, buka pengaturan server Discord Anda, klik "Integrations", lalu pilih "View Webhooks". Setelah itu, klik "New Webhook".

Ubah gambar tampilan, nama, dan kanal. Lalu salin URL Webhook.

URL webhook terdiri dari dua bagian, yaitu Webhook Id dan Webhook token: https://discord.com/api/webhooks/<webhook_id>/<webhook_token>. Untuk membuat notifikasi Prefect, kita perlu memisahkan kedua bagian ini secara manual dari tautan.

Untuk membuat notifikasi Discord di Prefect UI, buka opsi "Notifications" dan klik tombol "+" di bagian atas. Lalu, pilih status run yang diinginkan, tag, webhook Discord, dan tipe notifikasi.

Selanjutnya, masukkan Webhook ID dan Webhook token lalu klik tombol biru "Create".

Kita perlu mengeksekusi deployment agar notifikasi muncul di kanal utama server Discord kita.

$ prefect deployment run 'ml-workflow/ml_workflow_bank_churn'

Luar biasa. Dengan metode yang sama, kita dapat menyiapkan berbagai layanan pesan seperti Microsoft Teams, Slack, SMS, dan email, bahkan membuat webhook kustom sendiri.

Ikuti kursus Designing Machine Learning Workflows in Python untuk mempelajari cara membangun pipeline yang tahan uji waktu.

Menjalankan Deployment di Cloud

Setelah menerapkan flow machine learning secara lokal, saatnya menerapkannya di Prefect Cloud. Ini adalah proses yang jauh lebih mudah daripada deployment lokal, karena Anda tidak perlu memulai server atau agent, atau mengeksekusi run menggunakan alat CLI. Semuanya dikelola oleh Prefect.

Mendorong Kode ke Repositori GitHub

Sebelum menerapkan flow saat ini, penting untuk dicatat bahwa akan terjadi error karena mencoba mengakses file dan data dari direktori lokal yang tidak dapat diakses di cloud. Untuk membuat file tersebut tersedia bagi Prefect Cloud, kita harus membuat dan mendorong kode ke repositori GitHub publik.

Ini dapat dilakukan dengan mudah melalui situs web GitHub atau menggunakan ekstensi Git di VSCode untuk membuat repositori remote dan mendorong kode melalui antarmuka pengguna.

Berikut adalah repositori proyek yang akan kita gunakan untuk deployment cloud: kingabzpro/ML-Workflow-Orchestration-With-Prefect

Menyiapkan Prefect Cloud

Buat akun Prefect Cloud Anda dengan membuka https://app.prefect.cloud/auth/login lalu masuk menggunakan alat Prefect CLI.

$ prefect cloud login

Cukup pilih "Log in with a web browser" dan ikuti langkah-langkah untuk mengautentikasi klien lokal dengan Prefect Cloud.

Kita akan membuat work pool yang dikelola Prefect untuk sepenuhnya memanfaatkan infrastruktur Prefect Cloud alih-alih menggunakan agent lokal kita.

$ prefect work-pool create DC-work-pool --type prefect:managed

Untuk akun gratis, Prefect menyediakan 10 jam komputasi gratis per bulan yang dapat Anda gunakan untuk bereksperimen.

Deployment Kode Python

Sekarang kita akan menerapkan flow kita ke Prefect Cloud secara terprogram.

Buat file baru bernama deployment.py.

Siapkan argumen untuk fungsi flow Prefect:

• source: tautan ke repositori GitHub yang berisi file flow.
• entrypoint: nama file dan nama fungsi flow.
• work_pool_name: berikan nama work pool terkelola Prefect.
• tags: untuk pengkategorian.
• job_variables: untuk memasang paket Python yang diperlukan agar kode dapat berjalan. Jika Anda menggunakan work pool hibrida seperti agent Prefect kita, Anda tidak perlu memberikan job_variables.

from prefect import flow

if __name__ == "__main__":
    flow.from_source(
        source="https://github.com/kingabzpro/ML-Workflow-Orchestration-With-Prefect.git",
        entrypoint="main.py:ml_workflow",
    ).deploy(
        name="first-prefect-deployment",
        work_pool_name="DC-work-pool",
        tags=["dev"],
        job_variables={"pip_packages": ["pandas", "skops", "scikit-learn"]},
    )

Jalankan kode Python.

$ python deployment.py

Flow diterapkan di Prefect Cloud. Flow dapat dieksekusi melalui CLI atau dasbor Prefect Cloud.

Menjalankan Flow di Cloud

Untuk menjalankan flow, buka situs: https://app.prefect.cloud/. Setelah itu, buka bagian “Deployments” lalu klik tombol tiga titik vertikal pada deployment Anda, dan pilih opsi “Quick run”.

Untuk melihat bagaimana performa run kita, buka bagian “Flow Run” dan klik run terbaru untuk melihat grafik alur kerja.

Diagram flow menyajikan gambaran keterhubungan antar task dan waktu yang dibutuhkan tiap proses.

Untuk log terperinci, gulir ke bawah dan lihat semua task yang dijalankan beserta cap waktu.

Jika kita pergi ke bagian “Work Pools”, kita dapat melihat dengan jelas bahwa kita telah menggunakan 0,1 jam dari batas 10 jam untuk pengguna gratis.

Lihat panduan Machine Learning Workflow, yang menyertakan infografik untuk membantu Anda memahami task yang terlibat dalam penyiapan proyek, persiapan data, pemodelan, dan deployment.

Penutup

Menyiapkan satu alur kerja machine learning itu mudah. Dalam skenario dunia nyata, Anda mungkin perlu menjalankan banyak alur kerja machine learning dan data secara bersamaan. Di sinilah Prefect unggul. Prefect menawarkan skalabilitas dan fleksibilitas dalam memilih untuk menjalankannya di lingkungan lokal atau cloud untuk server Prefect atau work pool. Ini adalah alat yang fantastis untuk menjadwalkan, mengotomatiskan, dan mengelola alur kerja.

Dalam tutorial ini, kami telah membahas fitur dan fungsionalitas utama ekosistem Prefect. Kita juga telah mempelajari cara menerapkan flow machine learning secara lokal dan di cloud. Untuk melanjutkan perjalanan MLOps Anda, langkah berikutnya adalah mempelajari cara membangun CI/CD untuk alur kerja machine learning dengan mengikuti "A Beginner's Guide to CI/CD for Machine Learning."

Prefect hanyalah salah satu alat dalam bidang MLOps yang digunakan untuk orkestrasi alur kerja. Untuk mempelajari alat penting lainnya untuk versioning data, feature store, pengujian model, deployment dan serving model, pemantauan model, runtime engine, serta alat MLOps end-to-end, bacalah artikel 25 Top MLOps Tools You Need to Know in 2024. Anda juga dapat membaca perbandingan kami tentang Prefect vs Airflow untuk melihat perbandingan kedua alat tersebut.