Bagian I - Bab 5

Data Engineering: Pipeline dan Persiapan Data

Merancang pipeline ETL (Extract, Transform, Load) secara sistematis untuk menyiapkan data sebelum masuk ke sistem AI.

5.1 Konsep Pipeline ETL

Pipeline data dalam konteks AI Engineering mengikuti pola tiga tahap yang dikenal sebagai ETL: Extract, Transform, Load. Setiap tahap memiliki tanggung jawab yang jelas dan terpisah, ini bukan sekadar konvensi penamaan, tapi prinsip desain yang membuat pipeline mudah di-debug dan mudah dimodifikasi.

• Extract adalah proses mengambil data dari sumbernya, bisa berupa file CSV, endpoint API, database, atau dokumen PDF. Tahap ini hanya bertanggung jawab mengambil data apa adanya, tanpa memodifikasinya.
• Transform adalah tahap di mana data dibersihkan, dinormalisasi, dan diubah ke format yang dibutuhkan. Ini adalah tahap yang paling kompleks dan paling sering menjadi sumber masalah.
• Load adalah proses menyimpan data yang sudah bersih ke tujuan akhirnya, bisa berupa file CSV baru, database, atau vector database untuk sistem RAG.

Mengapa pemisahan ini penting? Karena ketika pipeline gagal, kamu bisa langsung tahu di tahap mana masalahnya terjadi tanpa harus menelusuri kode yang mencampur ketiga tanggung jawab tersebut dalam satu fungsi panjang.

5.2 Extract: Mengambil Data dari Berbagai Sumber

Sumber data yang paling umum ditemui dalam proyek AI Engineering adalah file lokal, API eksternal, dan dataset publik. Masing-masing memiliki cara pengambilan yang berbeda, tapi semuanya berujung pada satu tujuan yang sama: DataFrame Pandas yang siap diproses.

File Lokal

Membaca data dari file CSV, JSON, atau Excel yang tersimpan secara lokal. Format paling umum untuk data yang sudah disiapkan sebelumnya.

import pandas as pd

# CSV, format paling umum
df = pd.read_csv("data/produk.csv")

# CSV dengan separator berbeda (misal: semicolon)
df = pd.read_csv("data/ekspor.csv", sep=";", encoding="utf-8")

# JSON, satu objek per baris (JSON Lines format)
df = pd.read_json("data/log.jsonl", lines=True)

# Excel, jika data berasal dari spreadsheet
df = pd.read_excel("data/laporan.xlsx", sheet_name="Sheet1")

# Selalu verifikasi hasil setelah load
print(f"Shape: {df.shape}")   # (baris, kolom)
print(df.head(3))             # 3 baris pertama
print(df.dtypes)              # tipe data setiap kolom

API Eksternal

Mengambil data dari API eksternal menggunakan library requests. Cocok untuk data yang diperbarui secara berkala atau data yang tidak tersedia dalam bentuk file.

import requests
import pandas as pd
import os
from dotenv import load_dotenv

load_dotenv()

def extract_dari_api(endpoint: str, params: dict = None) -> pd.DataFrame:
    """
    Ambil data dari API endpoint dan kembalikan sebagai DataFrame.
    Pisahkan fungsi extract dari fungsi transform.
    """
    headers = {
        "Authorization": f"Bearer {os.getenv('API_KEY')}",
        "Content-Type": "application/json"
    }

    response = requests.get(endpoint, headers=headers, params=params)

    if response.status_code != 200:
        raise RuntimeError(
            f"API request gagal: {response.status_code} - {response.text}"
        )

    data = response.json()

    # Sesuaikan path navigasi JSON dengan struktur API yang digunakan
    records = data.get("data", data)  # fallback ke root jika tidak ada key "data"
    return pd.DataFrame(records)

# Penggunaan
df = extract_dari_api(
    endpoint="https://api.example.com/produk",
    params={"limit": 500, "kategori": "elektronik"}
)

Dataset Publik

Mengambil dataset dari Hugging Face atau Kaggle, sumber data publik yang paling umum digunakan dalam proyek AI. Dataset dari sini sering digunakan untuk fine-tuning atau evaluasi.

# Dari Hugging Face Datasets
from datasets import load_dataset

# Load dataset publik langsung ke memori
dataset = load_dataset("indonlp/indonesian-nlp", split="train")

# Konversi ke DataFrame Pandas untuk manipulasi lebih lanjut
df = dataset.to_pandas()
print(f"Dataset dimuat: {len(df)} baris")

# Dari Kaggle (setelah kaggle API dikonfigurasi)
import subprocess
import pandas as pd

# Download dataset via Kaggle CLI
subprocess.run([
    "kaggle", "datasets", "download",
    "-d", "username/nama-dataset",
    "--unzip", "-p", "data/"
])

df = pd.read_csv("data/nama-dataset.csv")

PDF / Dokumen

Mengekstrak teks dari file PDF, sumber data yang sangat umum dalam sistem RAG. Teks yang diekstrak ini akan menjadi input untuk proses chunking di Bab 13.

import pdfplumber
import pandas as pd

def extract_dari_pdf(path_file: str) -> pd.DataFrame:
    """
    Ekstrak teks dari setiap halaman PDF.
    Kembalikan sebagai DataFrame dengan kolom halaman dan konten.
    """
    records = []

    with pdfplumber.open(path_file) as pdf:
        for nomor, halaman in enumerate(pdf.pages, start=1):
            teks = halaman.extract_text()

            # Lewati halaman kosong atau yang tidak berisi teks
            if not teks or not teks.strip():
                continue

            records.append({
                "halaman": nomor,
                "konten": teks.strip(),
                "jumlah_karakter": len(teks.strip())
            })

    df = pd.DataFrame(records)
    print(f"Berhasil mengekstrak {len(df)} halaman dari {path_file}")
    return df

# Penggunaan
df_pdf = extract_dari_pdf("dokumen/sop_perusahaan.pdf")

5.3 Transform: Membersihkan dan Menyiapkan Data

Tahap Transform adalah yang paling menentukan kualitas output sistem AI. Data mentah hampir selalu mengandung masalah: nilai yang hilang, duplikat, tipe data yang tidak konsisten, atau format yang tidak seragam. Masalah-masalah ini, jika tidak ditangani, akan menyebar dan memperbesar kesalahan di tahap berikutnya.

Ada empat kategori masalah yang paling sering ditemui.

Missing Values

Dua strategi utama: hapus baris yang memiliki nilai kosong, atau isi dengan nilai pengganti. Pilihan bergantung pada seberapa penting baris tersebut dan apakah ada nilai yang masuk akal sebagai pengganti.

# Lihat berapa banyak nilai kosong per kolom
print(df.isnull().sum())

# Strategi 1: Hapus baris yang memiliki nilai kosong
# Gunakan ini jika baris tersebut tidak bisa diselamatkan
df = df.dropna()

# Hapus hanya jika kolom tertentu kosong
df = df.dropna(subset=["nama", "harga"])

# Strategi 2: Isi dengan nilai pengganti
# Untuk kolom numerik: isi dengan 0 atau nilai rata-rata
df["stok"] = df["stok"].fillna(0)
df["harga"] = df["harga"].fillna(df["harga"].mean())

# Untuk kolom teks: isi dengan label eksplisit
df["kategori"] = df["kategori"].fillna("tidak_diketahui")

Duplikat

Duplikat bisa muncul karena proses pengambilan data yang berulang atau penggabungan dataset dari beberapa sumber. Penting untuk diperiksa sebelum data dimasukkan ke sistem AI karena duplikat bisa membuat model terlalu bias pada data yang terduplikasi.

# Hitung jumlah baris duplikat
jumlah_duplikat = df.duplicated().sum()
print(f"Ditemukan {jumlah_duplikat} baris duplikat")

# Lihat baris mana saja yang duplikat
print(df[df.duplicated(keep=False)])

# Hapus duplikat, pertahankan kemunculan pertama
df = df.drop_duplicates()

# Hapus duplikat berdasarkan kolom tertentu saja
# Berguna jika ID produk harus unik meski kolom lain berbeda
df = df.drop_duplicates(subset=["id_produk"], keep="first")

print(f"Setelah deduplikasi: {len(df)} baris tersisa")

Tipe Data

Data dari CSV atau API sering kali memiliki tipe data yang tidak sesuai, angka tersimpan sebagai string, atau tanggal tersimpan sebagai teks biasa. Operasi matematis atau perbandingan akan gagal jika tipe data tidak benar.

# Periksa tipe data saat ini
print(df.dtypes)

# Konversi kolom ke tipe yang benar
df["harga"] = df["harga"].astype(float)
df["stok"] = df["stok"].astype(int)

# to_numeric lebih aman: kolom dengan nilai non-numerik
# akan diubah ke NaN, bukan menyebabkan error
df["harga"] = pd.to_numeric(df["harga"], errors="coerce")

# Konversi kolom tanggal dari string ke datetime
df["tanggal"] = pd.to_datetime(df["tanggal"], format="%Y-%m-%d")

# Hapus karakter non-numerik sebelum konversi
# Contoh: "Rp 50.000" -> 50000
df["harga"] = (
    df["harga"]
    .str.replace("Rp", "")
    .str.replace(".", "")
    .str.strip()
    .astype(float)
)

Format Teks

Inkonsistensi format teks adalah masalah yang sering diabaikan tapi berdampak besar pada sistem RAG. Jika nama kota tersimpan sebagai "Jakarta", "jakarta", dan "JAKARTA", sistem pencarian tidak akan menganggap ketiganya sama.

# Standarisasi kapitalisasi
df["kota"] = df["kota"].str.strip().str.title()
# "JAKARTA" -> "Jakarta", "jakarta" -> "Jakarta"

# Hapus spasi berlebih di dalam teks
df["deskripsi"] = df["deskripsi"].str.replace(r'\s+', ' ', regex=True)

# Normalisasi nilai kategorikal yang tidak konsisten
pemetaan_kategori = {
    "elektronik": "Elektronik",
    "Elektronik": "Elektronik",
    "elec": "Elektronik",
    "fashion": "Fashion",
    "Fashion": "Fashion"
}
df["kategori"] = df["kategori"].map(pemetaan_kategori)

# Hapus karakter khusus yang tidak diinginkan dari teks
df["konten"] = df["konten"].str.replace(r'[^\w\s.,!?]', '', regex=True)

Fungsi Transform Lengkap

Fungsi transform yang menggabungkan semua langkah pembersihan dalam satu unit yang bisa dipanggil ulang. Ini adalah pola yang direkomendasikan: satu fungsi, satu tanggung jawab, mudah diuji dan dimodifikasi.

import pandas as pd

def transform_data_produk(df: pd.DataFrame) -> pd.DataFrame:
    """
    Bersihkan dan standarisasi DataFrame produk.
    Input: DataFrame mentah dari sumber manapun.
    Output: DataFrame bersih siap digunakan model AI.
    """
    df = df.copy()  # jangan ubah DataFrame asli

    # 1. Standarisasi nama kolom
    df.columns = df.columns.str.strip().str.lower().str.replace(' ', '_')

    # 2. Hapus baris tanpa nama atau harga
    df = df.dropna(subset=["nama", "harga"])

    # 3. Konversi tipe data
    df["harga"] = pd.to_numeric(df["harga"], errors="coerce")
    df["stok"] = df["stok"].fillna(0).astype(int)

    # 4. Bersihkan teks
    df["nama"] = df["nama"].str.strip().str.title()
    df["deskripsi"] = df["deskripsi"].str.strip().fillna("")

    # 5. Hapus duplikat
    df = df.drop_duplicates(subset=["nama"], keep="first")

    # 6. Reset index setelah semua filter
    df = df.reset_index(drop=True)

    print(f"Transform selesai: {len(df)} baris bersih")
    return df

5.4 Load: Menyimpan Data ke Tujuan Akhir

Setelah data bersih, tahap terakhir adalah menyimpannya ke tujuan yang sesuai dengan kebutuhan sistem. Pilihan tujuan penyimpanan bergantung pada bagaimana data akan digunakan selanjutnya.

File Lokal

Menyimpan DataFrame bersih ke file lokal. Pilihan paling sederhana untuk data yang akan digunakan oleh pipeline berikutnya atau dibagikan ke rekan tim.

import os

def load_ke_file(df: pd.DataFrame, path_output: str) -> None:
    """
    Simpan DataFrame bersih ke file.
    Buat direktori output jika belum ada.
    """
    # Pastikan direktori tujuan sudah ada
    os.makedirs(os.path.dirname(path_output), exist_ok=True)

    if path_output.endswith(".csv"):
        # index=False: tidak simpan nomor baris sebagai kolom
        df.to_csv(path_output, index=False, encoding="utf-8")

    elif path_output.endswith(".json"):
        df.to_json(
            path_output,
            orient="records",   # satu objek per baris
            force_ascii=False,  # pertahankan karakter non-ASCII
            indent=2
        )

    elif path_output.endswith(".parquet"):
        # Parquet: format kolumnar yang efisien untuk dataset besar
        df.to_parquet(path_output, index=False)

    print(f"Data tersimpan ke: {path_output} ({len(df)} baris)")

# Penggunaan
load_ke_file(df_bersih, "data/processed/produk_bersih.csv")

Teks untuk RAG

Menyiapkan teks dari DataFrame untuk dimasukkan ke pipeline RAG. Output dari tahap ini adalah daftar string yang siap di-chunk dan di-embed di Bab 13.

from typing import List

def load_ke_teks_rag(df: pd.DataFrame, kolom_konten: str) -> List[str]:
    """
    Konversi DataFrame ke daftar string untuk pipeline RAG.
    Setiap string mewakili satu unit dokumen yang akan di-chunk.
    """
    # Filter baris dengan konten kosong
    df_valid = df[df[kolom_konten].str.strip().str.len() > 0].copy()

    # Jika tersedia metadata, gabungkan sebagai konteks
    # Ini membantu model memahami asal-usul setiap potongan teks
    if "judul" in df_valid.columns:
        dokumen = (
            "Judul: " + df_valid["judul"] + "\n\n" + df_valid[kolom_konten]
        ).tolist()
    else:
        dokumen = df_valid[kolom_konten].tolist()

    print(f"{len(dokumen)} dokumen siap untuk pipeline RAG")
    return dokumen

# Penggunaan
dokumen_rag = load_ke_teks_rag(df_bersih, kolom_konten="deskripsi")

5.5 Merakit Pipeline ETL Lengkap

Ketiga tahap di atas, Extract, Transform, Load, digabungkan menjadi satu fungsi pipeline yang bisa dijalankan ulang kapanpun dibutuhkan. Ini adalah pola yang paling penting dari seluruh bab ini: pipeline yang bisa dieksekusi dengan satu perintah, menghasilkan output yang konsisten setiap kali dijalankan.

Ada satu tahap tambahan yang tidak selalu disebut dalam definisi ETL tapi sangat penting dalam praktik: Validate, pemeriksaan singkat setelah Extract untuk memastikan data yang masuk sudah sesuai ekspektasi sebelum diproses lebih lanjut. Ini mencegah masalah yang baru ketahuan di akhir pipeline.

import pandas as pd
import requests, os
from dotenv import load_dotenv
from typing import List

load_dotenv()

# ============================================
# EXTRACT
# ============================================
def extract(sumber: str) -> pd.DataFrame:
    """Ambil data mentah dari sumber."""
    if sumber.endswith(".csv"):
        return pd.read_csv(sumber)
    elif sumber.startswith("http"):
        resp = requests.get(sumber, headers={
            "Authorization": f"Bearer {os.getenv('API_KEY')}"
        })
        resp.raise_for_status()
        return pd.DataFrame(resp.json().get("data", []))
    else:
        raise ValueError(f"Format sumber tidak dikenali: {sumber}")

# ============================================
# VALIDATE
# ============================================
def validate(df: pd.DataFrame, kolom_wajib: List[str]) -> None:
    """
    Periksa struktur dasar DataFrame sebelum diproses.
    Lempar error lebih awal daripada gagal diam-diam di tengah pipeline.
    """
    kolom_hilang = [k for k in kolom_wajib if k not in df.columns]
    if kolom_hilang:
        raise ValueError(f"Kolom wajib tidak ditemukan: {kolom_hilang}")
    if len(df) == 0:
        raise ValueError("DataFrame kosong, tidak ada data untuk diproses")
    print(f"Validasi lulus: {len(df)} baris, {len(df.columns)} kolom")

# ============================================
# TRANSFORM
# ============================================
def transform(df: pd.DataFrame) -> pd.DataFrame:
    """Bersihkan dan standarisasi data."""
    df = df.copy()
    df.columns = df.columns.str.strip().str.lower().str.replace(' ', '_')
    df = df.dropna(subset=["nama", "harga"])
    df["harga"] = pd.to_numeric(df["harga"], errors="coerce")
    df["stok"] = df["stok"].fillna(0).astype(int)
    df["nama"] = df["nama"].str.strip().str.title()
    df = df.drop_duplicates(subset=["nama"]).reset_index(drop=True)
    print(f"Transform selesai: {len(df)} baris bersih")
    return df

# ============================================
# LOAD
# ============================================
def load(df: pd.DataFrame, path_output: str) -> None:
    """Simpan data bersih ke tujuan akhir."""
    os.makedirs(os.path.dirname(path_output), exist_ok=True)
    df.to_csv(path_output, index=False, encoding="utf-8")
    print(f"Data tersimpan: {path_output}")

# ============================================
# PIPELINE, titik masuk tunggal
# ============================================
def run_pipeline(sumber: str, output: str) -> pd.DataFrame:
    """
    Jalankan pipeline ETL lengkap dari sumber ke output.
    Kembalikan DataFrame bersih untuk penggunaan langsung.
    """
    print(f"Memulai pipeline: {sumber}")

    df_mentah = extract(sumber)
    validate(df_mentah, kolom_wajib=["nama", "harga", "stok"])
    df_bersih = transform(df_mentah)
    load(df_bersih, output)

    print("Pipeline selesai")
    return df_bersih

# Jalankan dengan satu baris
df = run_pipeline(
    sumber="data/raw/produk.csv",
    output="data/processed/produk_bersih.csv"
)

Kesalahan Umum

Ringkasan Bab 5