Pipeline data yang baik dimulai dari pemisahan yang jelas antara apa yang dilakukan setiap stage dan bagaimana stage dieksekusi. Pada Bagian 1 ini kita mendesain pipeline dengan empat stage (ambil, bersihkan, transformasi, simpan) di mana setiap stage adalah object dengan pure function. Eksekusi masih sequential; concurrency akan ditambahkan di Bagian 2. Pendekatan ini membuat pipeline mudah diuji (setiap stage dapat diuji terpisah) dan fleksibel (cara eksekusi dapat diganti tanpa mengubah stage).

Pipeline Multi-Stage #

Pipeline terdiri dari empat stage yang masing-masing melakukan satu tugas spesifik:

  • Ambil: membaca data mentah dari sumber (file, database, API). Output: List<Record>.
  • Bersihkan: normalisasi data (uppercase, trim, replace). Output: List<Record> dengan field cleaned.
  • Transformasi: menambahkan field turunan (metadata, agregasi). Output: List<Record> dengan field transformed.
  • Simpan: menyimpan ke destination (database, file). Output: List<Record> dengan flag stored = true.

Setiap stage adalah pure function: input yang sama selalu menghasilkan output yang sama, tanpa efek samping. Pure function mudah diuji dan dapat dipanggil berulang tanpa khawatir state.

Model Record #

Model data pipeline adalah Record dengan field opsional yang diisi stage by stage:

kotlin Record model 
data class Record(
    val id: Int,
    val raw: String,
    val cleaned: String? = null,
    val transformed: String? = null,
    val stored: Boolean = false
)

Field nullable (cleaned, transformed) memungkinkan tracking stage mana yang sudah dijalankan. Stage Bersihkan mengisi cleaned; stage Transformasi mengisi transformed; stage Simpan mengatur stored = true. Jika suatu record belum diproses penuh, field yang belum diisi tetap null — memudahkan debugging.

Implementasi Pipeline Sequential #

Kode berikut mendemonstrasikan pipeline dengan empat stage pure function dan orchestrator yang merangkainya:

kotlin 
// Bagian 1: Desain pipeline data multi-stage.
// Mendemonstrasikan pipeline dengan stage yang merupakan pure function.
// Tidak ada concurrency di bagian ini; eksekusi sequential.

// === Stage 1: Ambil data mentah ===
data class Record(
    val id: Int,
    val raw: String,
    val cleaned: String? = null,
    val transformed: String? = null,
    val stored: Boolean = false
)

object AmbilStage {
    fun eksekusi(jumlah: Int): List<Record> {
        val hasil = mutableListOf<Record>()
        for (i in 1..jumlah) {
            hasil.add(Record(id = i, raw = "raw-$i"))
        }
        return hasil
    }
}

// === Stage 2: Bersihkan data ===
object BersihkanStage {
    fun eksekusi(r: Record): Record =
        r.copy(
            cleaned = r.raw.uppercase().replace("RAW-", "CLEAN-")
        )

    fun eksekusiSemua(records: List<Record>): List<Record> =
        records.map { eksekusi(it) }
}

// === Stage 3: Transformasi ===
object TransformasiStage {
    fun eksekusi(r: Record): Record =
        r.copy(
            transformed = "${r.cleaned}|step3|id=${r.id}"
        )

    fun eksekusiSemua(records: List<Record>): List<Record> =
        records.map { eksekusi(it) }
}

// === Stage 4: Simpan ===
object SimpanStage {
    fun eksekusi(r: Record): Record =
        r.copy(stored = true)

    fun eksekusiSemua(records: List<Record>): List<Record> =
        records.map { eksekusi(it) }
}

// === Pipeline orchestrator ===
object Pipeline {
    fun jalankan(jumlah: Int): List<Record> {
        // Stage 1: Ambil
        val r1 = AmbilStage.eksekusi(jumlah)

        // Stage 2: Bersihkan
        val r2 = BersihkanStage.eksekusiSemua(r1)

        // Stage 3: Transformasi
        val r3 = TransformasiStage.eksekusiSemua(r2)

        // Stage 4: Simpan
        val r4 = SimpanStage.eksekusiSemua(r3)

        return r4
    }
}

// === Pipeline builder pattern ===
class PipelineBuilder<T> {
    private val stages = mutableListOf<(T) -> T>()

    fun stage(s: (T) -> T): PipelineBuilder<T> {
        stages.add(s)
        return this
    }

    fun eksekusi(input: T): T {
        var current = input
        for (s in stages) {
            current = s(current)
        }
        return current
    }

    fun eksekusiSemua(inputs: List<T>): List<T> =
        inputs.map { eksekusi(it) }
}

fun main() {
    println("=== Pipeline Data Multi-Stage ===")
    println("Stage: Ambil -> Bersihkan -> Transformasi -> Simpan\n")

    println("=== Eksekusi pipeline (5 record) ===")
    val hasil = Pipeline.jalankan(5)
    hasil.forEach { r ->
        println("  Record #${r.id} raw=${r.raw} cleaned=${r.cleaned} transformed=${r.transformed} stored=${r.stored}")
    }

    println()
    println("=== Pipeline builder pattern ===")
    val builder = PipelineBuilder<Record>()
        .stage { it.copy(cleaned = it.raw.uppercase()) }
        .stage { it.copy(transformed = "${it.cleaned}|step3") }
        .stage { it.copy(stored = true) }

    val input = Record(99, "raw-99")
    val output = builder.eksekusi(input)
    println("  Input : $input")
    println("  Output: $output")

    println()
    println("=== Stage dijalankan terpisah ===")
    val r = AmbilStage.eksekusi(1).first()
    println("  Setelah Ambil       : $r")
    val r2 = BersihkanStage.eksekusi(r)
    println("  Setelah Bersihkan   : $r2")
    val r3 = TransformasiStage.eksekusi(r2)
    println("  Setelah Transformasi: $r3")
    val r4 = SimpanStage.eksekusi(r3)
    println("  Setelah Simpan      : $r4")
}

Output yang Dihasilkan:

text 
=== Pipeline Data Multi-Stage ===
Stage: Ambil -> Bersihkan -> Transformasi -> Simpan

=== Eksekusi pipeline (5 record) ===
  Record #1 raw=raw-1 cleaned=CLEAN-1 transformed=CLEAN-1|step3|id=1 stored=true
  Record #2 raw=raw-2 cleaned=CLEAN-2 transformed=CLEAN-2|step3|id=2 stored=true
  Record #3 raw=raw-3 cleaned=CLEAN-3 transformed=CLEAN-3|step3|id=3 stored=true
  Record #4 raw=raw-4 cleaned=CLEAN-4 transformed=CLEAN-4|step3|id=4 stored=true
  Record #5 raw=raw-5 cleaned=CLEAN-5 transformed=CLEAN-5|step3|id=5 stored=true

=== Pipeline builder pattern ===
  Input : Record(id=99, raw=raw-99, cleaned=null, transformed=null, stored=false)
  Output: Record(id=99, raw=raw-99, cleaned=RAW-99, transformed=RAW-99|step3, stored=true)

=== Stage dijalankan terpisah ===
  Setelah Ambil       : Record(id=1, raw=raw-1, cleaned=null, transformed=null, stored=false)
  Setelah Bersihkan   : Record(id=1, raw=raw-1, cleaned=CLEAN-1, transformed=null, stored=false)
  Setelah Transformasi: Record(id=1, raw=raw-1, cleaned=CLEAN-1, transformed=CLEAN-1|step3|id=1, stored=false)
  Setelah Simpan      : Record(id=1, raw=raw-1, cleaned=CLEAN-1, transformed=CLEAN-1|step3|id=1, stored=true)

Mari kita bedah komponen utama:

  1. data class Record dengan field nullable: field cleaned, transformed nullable memungkinkan tracking stage mana yang sudah dijalankan. Saat Record baru dibuat di stage Ambil, semua field opsi null. Stage berikutnya mengisi field melalui copy().
  2. object BersihkanStage { fun eksekusi(r: Record): Record = r.copy(cleaned = ...) }: stage sebagai pure function. Input Record, output Record baru dengan field diisi. Tidak ada efek samping (no mutation, no I/O).
  3. records.map { eksekusi(it) }: mengaplikasikan stage ke setiap record. map adalah fungsi koleksi Kotlin yang mengembalikan list baru dengan hasil transformasi; tidak mengubah list asli.
  4. object Pipeline.jalankan(jumlah): orchestrator yang merangkai empat stage secara sequential. Output stage 1 menjadi input stage 2; output stage 2 menjadi input stage 3; dan seterusnya.
  5. PipelineBuilder<T> dengan method stage: builder pattern untuk merangkai stage secara dinamis. Method stage menambah lambda ke list; eksekusi menjalankan semua lambda secara berurutan. Builder fleksibel untuk pipeline yang berubah saat runtime.

Pure Function dan Testabilitas #

Stage di atas adalah pure function — input sama selalu menghasilkan output sama. Pure function mudah diuji:

kotlin Test pure function stage 
@Test
fun `Bersihkan mengubah raw-X menjadi CLEAN-X uppercase`() {
    val input = Record(id = 1, raw = "raw-1")
    val hasil = BersihkanStage.eksekusi(input)
    assertEquals("CLEAN-1", hasil.cleaned)
    assertNull(hasil.transformed)  // stage lain belum jalan
}

Test ini tidak butuh mock database atau fake I/O karena stage tidak melakukan I/O. Saat test pipeline penuh, kita dapat memalsukan stage Ambil dan Simpan dengan stub; stage Bersihkan dan Transformasi tetap pakai implementasi asli.

Builder Pattern untuk Pipeline Fleksibel #

PipelineBuilder<T> memungkinkan pipeline dirakit saat runtime:

kotlin Pipeline fleksibel 
val pipeline = PipelineBuilder<Record>()
    .stage(::bersihkan)
    .stage(::validasi)
    .stage(::enrich)
    .stage(::simpan)

Berbeda dengan object Pipeline yang hardcode empat stage, builder menerima lambda sehingga stage dapat dipilih berdasarkan konfigurasi. Berguna saat pipeline berbeda untuk input berbeda (misalnya, pipeline untuk file CSV berbeda dengan pipeline untuk JSON).

Referensi Lanjutan #

  • Kotlin coroutines design — dokumentasi desain coroutines untuk merancang pipeline data asinkron bertahap
  • Kotlin Sequences — dokumentasi Sequences sebagai dasar pemrosesan data lazy di tiap stage pipeline

Ringkasan Bagian 1 #

Pipeline multi-stage kini siap dengan empat pure function: AmbilStage, BersihkanStage, TransformasiStage, SimpanStage. Pipeline.jalankan() merangkai mereka secara sequential. PipelineBuilder memungkinkan pipeline fleksibel. Pada Bagian 2 kita menambahkan concurrency dengan coroutines: async/awaitAll untuk batch paralel, dan Flow untuk streaming record saat tersedia.