MLOps Pipeline Freelance IT Career — สร้างอาชีพ MLOps Freelance

MLOps Freelance

MLOps Pipeline Freelance IT Career ML Development Operations Data Pipeline Feature Engineering Model Training Deployment Monitoring Retraining MLflow Kubeflow

Stage	Tools	Input	Output	Frequency
Data Ingestion	Airflow Spark	Raw Data	Clean Dataset	Daily/Hourly
Feature Store	Feast Tecton	Clean Data	Features	On-demand
Training	PyTorch MLflow	Features	Model	On trigger
Evaluation	MLflow Great Expectations	Model + Test	Metrics	After training
Deployment	Seldon BentoML	Model	API Endpoint	On approval
Monitoring	Evidently Prometheus	Predictions	Alerts	Real-time

MLOps Pipeline

=== MLOps Pipeline with MLflow ===

อ่านเพิ่ม: LocalAI Self-hosted Event Driven Design | SiamCafe Blog · อ่านเพิ่ม: LLM Inference vLLM CQRS Event Sourcing | SiamCafe Blog · อ่านเพิ่ม: deep learning vs machine learning คือ — วิธีตั้งค่าและใช้งาน

pip install mlflow scikit-learn pandas

import mlflow

import mlflow.sklearn

from sklearn.ensemble import RandomForestClassifier

from sklearn.model_selection import train_test_split

from sklearn.metrics import accuracy_score, f1_score

import numpy as np

MLflow Experiment Tracking

mlflow.set_tracking_uri("http://mlflow-server:5000")

mlflow.set_experiment("customer-churn")

with mlflow.start_run(run_name="rf-baseline"):

# Log parameters

params = {"n_estimators": 100, "max_depth": 10, "min_samples_split": 5}

mlflow.log_params(params)

# Train model

model = RandomForestClassifier(**params)

model.fit(X_train, y_train)

# Evaluate

y_pred = model.predict(X_test)

metrics = {

"accuracy": accuracy_score(y_test, y_pred),

"f1_score": f1_score(y_test, y_pred),

}

mlflow.log_metrics(metrics)

# Log model

mlflow.sklearn.log_model(model, "model",

registered_model_name="churn-predictor")

print(f"Accuracy: {metrics['accuracy']:.4f}")

print(f"F1 Score: {metrics['f1_score']:.4f}")

DVC — Data Version Control

dvc init

dvc remote add -d s3remote s3://my-bucket/dvc-store

dvc add data/training.csv

git add data/training.csv.dvc .gitignore

git commit -m "Add training data v1"

dvc push

Airflow DAG — ML Pipeline

from airflow import DAG

from airflow.operators.python import PythonOperator

from datetime import datetime

dag = DAG('ml_pipeline', schedule_interval='@daily',

start_date=datetime(2024, 1, 1))

extract = PythonOperator(task_id='extract_data', python_callable=extract_fn, dag=dag)

transform = PythonOperator(task_id='transform_features', python_callable=transform_fn, dag=dag)

train = PythonOperator(task_id='train_model', python_callable=train_fn, dag=dag)

evaluate = PythonOperator(task_id='evaluate_model', python_callable=evaluate_fn, dag=dag)

deploy = PythonOperator(task_id='deploy_model', python_callable=deploy_fn, dag=dag)

extract >> transform >> train >> evaluate >> deploy

from dataclasses import dataclass

@dataclass

class PipelineStage:

stage: str

tool: str

automation: str

time: str

output: str

stages = [

PipelineStage("Data Ingestion", "Airflow + Spark", "Scheduled daily", "30 min", "Clean dataset S3"),

PipelineStage("Feature Engineering", "Feast Feature Store", "On-demand compute", "10 min", "Feature vectors"),

PipelineStage("Model Training", "PyTorch + MLflow", "Triggered on data change", "2-8 hours", "Model artifacts"),

PipelineStage("Evaluation", "MLflow + Great Expectations", "After every training", "5 min", "Metrics report"),

PipelineStage("Model Registry", "MLflow Registry", "Auto on improved metrics", "1 min", "Registered model"),

PipelineStage("Deployment", "BentoML + K8s", "On approval / auto", "10 min", "API endpoint"),

PipelineStage("Monitoring", "Evidently + Prometheus", "Real-time", "Continuous", "Dashboard alerts"),

]

print("=== MLOps Pipeline ===")

for s in stages:

print(f" [{s.stage}] Tool: {s.tool}")

print(f" Automation: {s.automation} | Time: {s.time}")

print(f" Output: {s.output}")

Model Serving

=== Model Serving with BentoML ===

pip install bentoml

BentoML Service

import bentoml

from bentoml.io import JSON, NumpyNdarray

model_ref = bentoml.mlflow.get("churn-predictor:latest")

model_runner = model_ref.to_runner()

svc = bentoml.Service("churn-service", runners=[model_runner])

@svc.api(input=JSON(), output=JSON())

async def predict(input_data: dict) -> dict:

features = preprocess(input_data)

prediction = await model_runner.predict.async_run(features)

return {

"prediction": int(prediction[0]),

"probability": float(prediction[0]),

"model_version": model_ref.tag.version,

}

Dockerfile — BentoML

bentoml build

bentoml containerize churn-service:latest

Kubernetes Deployment

apiVersion: apps/v1

kind: Deployment

spec:

replicas: 3

template:

spec:

containers:

• name: churn-service

image: churn-service:v1.0

resources:

requests: { cpu: "500m", memory: "1Gi" }

limits: { cpu: "2", memory: "4Gi" }

ports:

• containerPort: 3000

@dataclass

class ServingOption:

platform: str

latency: str

throughput: str

scale: str

cost: str

use_case: str

options = [

ServingOption("BentoML + K8s", "10-50ms", "1000+ rps", "HPA Auto", "$$", "General Production"),

ServingOption("Seldon Core", "15-100ms", "500+ rps", "K8s Native", "$$$", "Enterprise ML"),

ServingOption("TF Serving", "5-20ms", "2000+ rps", "K8s / Docker", "$$", "TensorFlow Models"),

ServingOption("Triton Server", "2-10ms", "5000+ rps", "GPU Optimized", "$$$", "GPU Inference"),

ServingOption("FastAPI + Docker", "10-30ms", "500+ rps", "Manual / K8s", "$", "Simple Deploy"),

ServingOption("SageMaker", "20-100ms", "Auto", "AWS Managed", "$$$", "AWS Native"),

]

print("\n=== Model Serving Options ===")

for o in options:

print(f" [{o.platform}] Latency: {o.latency} | Throughput: {o.throughput}")

print(f" Scale: {o.scale} | Cost: {o.cost} | Use: {o.use_case}")

Freelance Career

# === MLOps Freelance Career Path ===

@dataclass
class FreelanceGig:
 project_type: str
 duration: str
 rate: str
 skills_needed: str
 platform: str

gigs = [
 FreelanceGig("ML Pipeline Setup", "2-4 weeks", "$150-200/hr", "Airflow MLflow Docker K8s", "Toptal Upwork"),
 FreelanceGig("Model Deployment", "1-2 weeks", "$120-180/hr", "Docker K8s BentoML Seldon", "Toptal"),
 FreelanceGig("MLOps Consulting", "Ongoing", "$200-300/hr", "Architecture Strategy Best Practices", "Direct Client"),
 FreelanceGig("Feature Store Setup", "2-3 weeks", "$150-200/hr", "Feast Tecton Redis", "Upwork"),
 FreelanceGig("ML Monitoring", "1-2 weeks", "$120-160/hr", "Evidently Prometheus Grafana", "Toptal"),
 FreelanceGig("Data Pipeline", "3-6 weeks", "$100-150/hr", "Spark Airflow dbt", "Upwork Fiverr Pro"),
]

print("=== Freelance MLOps Gigs ===")
for g in gigs:
 print(f" [{g.project_type}] Duration: {g.duration}")
 print(f" Rate: {g.rate} | Skills: {g.skills_needed}")
 print(f" Platform: {g.platform}")

career_path = {
 "Year 1": "เรียน Python ML Docker เริ่ม Upwork $50-80/hr",
 "Year 2": "เพิ่ม K8s Cloud MLflow สร้าง Portfolio $80-120/hr",
 "Year 3": "Toptal Freelance สร้าง Blog Community $120-180/hr",
 "Year 4+": "Consulting Direct Client สอน Workshop $180-300/hr",
}

certifications = [
 "AWS Machine Learning Specialty",
 "Google Professional ML Engineer",
 "Databricks ML Associate",
 "CKA (Certified Kubernetes Admin)",
 "Terraform Associate",
]

print(f"\n\nCareer Path:")
for year, desc in career_path.items():
 print(f" [{year}]: {desc}")

print(f"\n Certifications:")
for c in certifications:
 print(f" - {c}")

เคล็ดลับ

• Portfolio: สร้าง End-to-end ML Pipeline บน GitHub
• Blog: เขียน Blog แชร์ MLOps Knowledge สร้างชื่อ
• Niche: เชี่ยวชาญ Niche เช่น NLP CV Recommender
• Community: เข้าร่วม MLOps Community Meetup Conference
• Automate: ทุก Pipeline ต้อง Reproducible และ Automated

MLOps คืออะไร

ML + Operations DevOps สำหรับ ML Data Pipeline Feature Training Deployment Monitoring Retraining MLflow Kubeflow Airflow Reproducible Automate

สรุป

MLOps Pipeline Freelance IT Career ML Training Deployment Monitoring MLflow BentoML Kubernetes Airflow Feature Store Portfolio Consulting Production