Belajar Apache Spark - Cross-System Integration

#Pendahuluan

Spark tidak pernah bekerja sendirian. Di arsitektur data nyata, Spark berdiri di tengah ekosistem: membaca dari Kafka, menulis ke Cassandra, mengirim dokumen ke Elasticsearch, dan mengisi data warehouse. Episode 18 ini membahas bagaimana Spark terhubung dengan sistem-sistem tersebut.

Kemampuan integrasi inilah yang menjadikan Spark "lem dalam data engineering". Seorang engineer yang menguasai integrasi bisa merancang aliran data yang utuh — dari sumber, melalui Spark, sampai ke konsumen — tanpa bergantung pada tool khusus untuk setiap pasangan sistem.

Episode ini membahas empat topik: konektor ke Kafka, Cassandra, dan Elasticsearch; Spark sebagai engine ETL untuk data lake dan warehouse; integrasi dengan BI tools; serta pola data ingestion dan CDC.

#Menghubungkan Spark dengan Kafka, Cassandra, dan Elasticsearch

#Kafka: Tulang Punggung Streaming

Kafka adalah sistem message queue paling umum di ekosistem Spark. Sebagai sumber, Spark membaca lewat Structured Streaming seperti di episode 10. Sebagai sink, hasil streaming dikirim kembali ke topic:

Membaca dan menulis Kafka

from pyspark.sql import functions as F
 
kafka_in = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "orders") \
    .load()
 
hasil = kafka_in.selectExpr("CAST(value AS STRING) AS value")
 
kafka_out = hasil.writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "orders-enriched") \
    .option("checkpointLocation", "data/ckpt-kafka") \
    .start()

selectExpr("CAST(value AS STRING) AS value") memastikan kolom value berformat string sebelum ditulis kembali. Kafka menyediakan jaminan exactly-once bila dikombinasikan dengan checkpoint yang benar.

#Cassandra: Menulis dengan Cepat

Cassandra adalah database distributed berbasis wide-column yang populer untuk workload write-heavy. Spark berintegrasi lewat connector DataStax:

Menulis ke Cassandra

df.write \
    .format("org.apache.spark.sql.cassandra") \
    .option("keyspace", "analitik") \
    .option("table", "ringkasan") \
    .mode("append") \
    .save()

df.write.format("org.apache.spark.sql.cassandra") menulis batch ke Cassandra. Connector ini mengatur token range dan batch size secara otomatis — penting diingat bahwa performa sangat bergantung pada desain primary key dan jumlah partition di tabel Cassandra.

#Elasticsearch: Index untuk Pencarian

Elasticsearch dipakai untuk pencarian teks dan dashboard. Spark mengirim dokumen lewat connector:

Mengirim ke Elasticsearch

df.write \
    .format("org.elasticsearch.spark.sql") \
    .option("es.nodes", "es-node:9200") \
    .option("es.resource", "analitik") \
    .mode("overwrite") \
    .save()

es.resource menentukan index tujuan. Untuk pipeline streaming, sink Elasticsearch juga didukung melalui connector yang sama.

#Spark sebagai ETL Engine untuk Data Lake dan Warehouse

#ETL ke Data Lake

Peran paling klasik: Spark menarik data dari berbagai sumber, membersihkan dan mentransformasi, lalu menulis ke data lake dalam format Parquet atau Delta:

sumber (DB, API, Kafka) → Spark (extract + transform) → data lake (Parquet/Delta)

Keunggulan Spark untuk ETL: satu engine untuk batch dan streaming, konektor ke hampir semua sumber, dan transformasi yang bisa diuji ulang.

#Memuat ke Data Warehouse

Untuk warehouse modern seperti Snowflake atau BigQuery, Spark menulis lewat konektor resmi:

Menulis ke Snowflake

df.write \
    .format("net.snowflake.spark.snowflake") \
    .option("sfUrl", "...") \
    .option("sfUser", "...") \
    .option("sfDatabase", "dw") \
    .option("sfSchema", "public") \
    .save()

Alternatif yang lebih ringan: Spark menulis Parquet ke staging di object storage, lalu warehouse melakukan COPY INTO — pola yang mengurangi beban konektor langsung dan memanfaatkan kecepatan native warehouse.

#Integrasi dengan BI Tools dan Analytics Platforms

#Menghubungkan BI ke Hasil Spark

BI tools tidak membaca data Spark secara langsung. Alur yang umum:

• JDBC/Thrift: jalankan Spark Thrift Server sehingga tool BI bisa mengquery data Spark dengan SQL standar.
• Konektor warehouse: BI membaca dari warehouse yang sudah diisi Spark.
• Format terbuka: BI membaca Parquet/Delta langsung dari data lake lewat engine seperti DuckDB atau Trino.

/opt/spark/sbin/start-thriftserver.sh

start-thriftserver.sh menyalakan endpoint JDBC/ODBC di port 10000 — tool seperti Tableau atau Superset lalu bisa terhubung seperti ke database biasa.

#Pola Pengelolaan Akses

Untuk skala besar, akses BI sebaiknya lewat warehouse atau engine terbuka, bukan lewat Spark session ad-hoc — supaya resource terisolasi dan tidak ada analis yang tidak sengaja menjalankan query raksasa di cluster produksi.

#Data Ingestion dan CDC Patterns

#CDC dengan Change Data Capture

CDC (Change Data Capture) menangkap perubahan di database (insert, update, delete) dan mengalirkannya untuk diproses. Pola umum:

1. Database mengeluarkan perubahan lewat binlog/WAL ke Kafka.
2. Spark membaca aliran perubahan dengan Structured Streaming.
3. Spark menerapkan perubahan ke target (data lake atau warehouse) dengan MERGE.

Terapkan CDC dengan merge

from pyspark.sql import functions as F
 
perubahan = spark.readStream.format("kafka").option("subscribe", "db-changes").load()
target = spark.read.format("delta").load("data/target_delta")
 
def proses_batch(df, epoch_id):
    df.createOrReplaceTempView("perubahan")
    target.sparkSession.sql("""
        MERGE INTO data/target_delta AS t
        USING perubahan AS s ON t.id = s.id
        WHEN MATCHED AND s.op = 'delete' THEN DELETE
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)
 
query = perubahan.writeStream.foreachBatch(proses_batch).start()

foreachBatch(proses_batch) memungkinkan logika batch penuh (termasuk MERGE) diterapkan pada setiap micro-batch — pola paling fleksibel untuk CDC di Spark.

#Idempotency dan Ordering

Perubahan database punya urutan yang penting. Saat menerapkan CDC, perhatikan:

• Idempotency: menerapkan perubahan yang sama dua kali harus menghasilkan hasil yang sama.
• Ordering: pastikan event untuk satu kunci diproses berurutan — grupkan per kunci dan beri timestamp.
• Exactly-once: gunakan checkpoint dan sink transaksional seperti Delta.

#Penutup

Episode 18 membekali kalian kemampuan integrasi: konektor Kafka, Cassandra, dan Elasticsearch menghubungkan Spark dengan sistem konsumen, peran Spark sebagai engine ETL mengisi data lake dan warehouse, BI tools menjangkau data lewat Thrift Server atau warehouse, dan pola CDC mempertahankan data tetap segar di target.

Inti yang harus dibawa pulang:

• Kafka sebagai sumber dan sink streaming dengan jaminan exactly-once.
• Cassandra dan Elasticsearch diakses lewat connector khusus format.
• Spark mengisi data lake dan warehouse; warehouse menyajikan ke BI.
• Thrift Server membuka SQL Spark ke tool BI standar.
• CDC mengalirkan perubahan database dan diproses dengan merge idempoten.

Di episode 19 selanjutnya kita akan membahas operational readiness dan runbooks — menyusun runbook untuk job failure dan recovery, incident response untuk cluster dan data corruption, backup konfigurasi dan artifacts, serta chaos testing untuk menguji ketahanan pipeline.