Integrasi File & Object Storage #

Dalam arsitektur pipa data modern, data yang mengalir melalui Apache Kafka sering kali memiliki dua tujuan utama: dikonsumsi langsung untuk kebutuhan real-time, dan disimpan ke dalam penyimpanan jangka panjang untuk analisis data besar (big data analytics) atau pengarsipan (archiving). Cloud Object Storage seperti Amazon S3, Google Cloud Storage (GCS), atau Azure Blob Storage adalah media penyimpanan paling populer karena biayanya yang sangat murah, kapasitasnya yang tidak terbatas, dan durabilitasnya yang sangat tinggi. Namun, memindahkan data dari Kafka ke object storage bukanlah sekadar menulis pesan teks biasa ke dalam folder. Kita harus memikirkan bagaimana data diatur agar efisien saat dibaca oleh query engine (seperti AWS Athena atau Snowflake), bagaimana mengompresi data, serta bagaimana mengatasi masalah file kecil (small file problem) yang dapat merusak kinerja metadata storage. Artikel ini akan membedah strategi integrasi file dan object storage secara komprehensif.

Penggunaan Utama Cloud Object Storage (S3 / GCS) #

Sebelum masuk ke aspek teknikal, mari kita bahas mengapa kita perlu menghubungkan Kafka dengan Cloud Object Storage menggunakan Kafka Connect:

1. Fondasi Data Lake dan Lakehouse #

Object storage bertindak sebagai repositori pusat (landing zone) di mana semua data mentah dari berbagai topik Kafka dikumpulkan sebelum diproses lebih lanjut. Data ini nantinya dapat dibaca secara paralel oleh query engine seperti Amazon Athena, Presto, Trino, Databricks, Apache Spark, atau Snowflake untuk kebutuhan business intelligence (BI) dan machine learning (ML).

2. Pengarsipan Data Jangka Panjang (Cold Storage Log Archiving) #

Kafka dirancang untuk menyimpan data aliran secara cepat dan dinamis, bukan untuk menyimpan data historis selama bertahun-tahun (karena biaya penyimpanan disk SSD pada broker Kafka sangat mahal). Dengan mengalirkan data ke S3 atau GCS, kita dapat menerapkan kebijakan penyimpanan data dingin (cold tier) yang sangat murah dan menyimpan data untuk kepatuhan regulasi (compliance audit trail) hingga bertahun-tahun.

3. Pemulihan Bencana (Disaster Recovery) #

Menyimpan salinan seluruh event Kafka ke object storage eksternal memberikan lapisan keamanan tambahan. Jika seluruh kluster Kafka kita mengalami kegagalan total yang tidak dapat dipulihkan, kita tetap dapat merekonstruksi keadaan data historis dengan memutar kembali (replay) data dari S3.

Masalah File Kecil (Small File Problem) dan Dampak Kritisnya #

Salah satu tantangan terbesar saat menulis data dari sistem streaming (seperti Kafka) ke sistem penyimpanan berbasis file (seperti S3) adalah Small File Problem (Masalah File Kecil).

Apa Itu Small File Problem? #

Jika kita menulis data ke S3 setiap kali ada pesan baru masuk ke Kafka, atau menggunakan interval waktu yang terlalu singkat (misalnya setiap 5 detik), kita akan menghasilkan jutaan file berukuran sangat kecil (misalnya berkisar antara 2 KB hingga 50 KB).

Masalah ini memiliki dampak operasional yang sangat buruk:

  • Overhead Biaya HTTP Request: Cloud provider seperti AWS mengenakan biaya berdasarkan jumlah request HTTP PUT (saat menulis) dan GET (saat membaca). Menulis jutaan file kecil akan melipatgandakan tagihan bulanan kita secara dramatis.
  • Performa Kueri Hancur: Ketika query engine seperti AWS Athena atau Spark membaca data untuk kebutuhan analisis, engine tersebut harus membuka, membaca, dan menutup ratusan ribu file kecil. Waktu yang terbuang untuk negosiasi jaringan HTTP dan pemrosesan metadata file jauh lebih besar daripada waktu pembacaan data aktual. Kueri yang seharusnya selesai dalam 3 detik bisa membengkak menjadi 30 menit.
  • Beban Metadata Engine: Sistem file terdistribusi (seperti HDFS) menyimpan metadata file di memori NameNode. Terlalu banyak file kecil akan membuat NameNode kehabisan memori (OutOfMemory).
flowchart TD
    subgraph BadApproach ["Skenario Buruk: File-File Kecil (High Latency & High Cost)"]
        direction TB
        KafkaMentah1[("Kafka Topics")] -->|Setiap 5 detik / 10 Record| S3Sink1["S3 Sink Task"]
        S3Sink1 -->|Tulis HTTP PUT| S3Bucket1[/"AWS S3 Bucket"\]
        S3Bucket1 --> F1["file1.json (5 KB)"]
        S3Bucket1 --> F2["file2.json (8 KB)"]
        S3Bucket1 --> F3["file3.json (4 KB)"]
        F1 & F2 & F3 -->|Query Lambat| Athena1["AWS Athena Query Engine"]
    end
    
    subgraph GoodApproach ["Skenario Baik: Rotasi Terkendali (Low Cost & High Performance)"]
        direction TB
        KafkaMentah2[("Kafka Topics")] -->|Buffer: 100.000 Record / 15 Menit| S3Sink2["S3 Sink Task"]
        S3Sink2 -->|Tulis HTTP PUT Sekaligus| S3Bucket2[/"AWS S3 Bucket"\]
        S3Bucket2 --> LargeFile["consolidated_file.parquet (128 MB)"]
        LargeFile -->|Query Cepat| Athena2["AWS Athena Query Engine"]
    end

Solusi: Mengendalikan Parameter Rotasi File #

Untuk menghindari masalah file kecil, kita harus memaksa Kafka Connect Sink to menahan (buffer) data di memori worker terlebih dahulu dan mengonsolidasikannya menjadi satu file besar (ukuran ideal berkisar antara 64 MB hingga 256 MB) sebelum dikirim ke S3.

Kita dapat mengontrol rotasi file menggunakan tiga parameter berikut di konfigurasi Sink Connector:

  1. flush.size: Jumlah minimum baris data (record) yang harus dikumpulkan di memori sebelum ditulis ke S3. (Contoh: setel ke 100000 record).
  2. rotate.interval.ms: Durasi waktu maksimal (dalam milidetik) sebelum file dirotasi berdasarkan waktu kedatangan data pertama dalam batch tersebut. (Contoh: 1200000 ms atau 20 menit).
  3. rotate.schedule.interval.ms: Rotasi file berdasarkan waktu dinding jam (wall-clock time) pada mesin worker Connect. Sangat berguna jika kita ingin file dirotasi tepat pada menit ke-00 setiap jam (misalnya pukul 13:00, 14:00) agar folder partisi bersih.

Proses rotasi file akan dipicu oleh parameter mana saja yang tercapai lebih dulu (whichever comes first).

Mekanisme Pemulihan Kesalahan saat Menulis ke S3 (Fault Tolerance on Uploads) #

Proses penulisan file biner berukuran besar ke cloud object storage melahirkan tantangan tersendiri terkait keandalan jaringan. Jika koneksi HTTP terputus saat file berukuran 100 MB sedang diunggah 90%, mengunggah ulang file dari awal adalah pemborosan resource yang besar.

1. Pemanfaatan S3 Multipart Uploads #

Untuk menjamin ketahanan unggahan, S3 Sink Connector secara otomatis membagi file besar menjadi bagian-bagian kecil (biasanya berukuran minimal 5 MB) dan mengunggahnya secara paralel menggunakan fitur Multipart Upload S3.

  • Parameter s3.part.size mendefinisikan ukuran minimum per segmen unggahan (default: 26214400 byte atau 25 MB). Jika file kita mencapai 100 MB, Connect akan mengunggahnya dalam 4 bagian paralel.
  • Jika salah satu segmen gagal diunggah karena timeout jaringan transien, worker Connect hanya akan mencoba mengunggah ulang segmen yang bermasalah tersebut, bukan seluruh file.

2. Penanganan Kebocoran Penyimpanan S3 (Aborted Multipart Uploads) #

Jika task Connect mengalami crash permanen di tengah-tengah proses unggahan multi-segmen, bagian file yang sudah terunggah sebagian akan tetap disimpan di S3 secara menggantung. S3 akan terus menagih biaya penyimpanan untuk segmen-segmen yatim (orphaned parts) ini meskipun file tersebut tidak pernah terkompilasi utuh.

  • Solusi Terbaik: Kita harus mengonfigurasi S3 Lifecycle Rules pada bucket S3 kita untuk secara otomatis menghapus multipart upload yang tidak selesai setelah lewat dari 7 hari: 
    {
  "Rules": [
    {
      "ID": "Hapus Multipart Upload yang Tertunda",
      "Status": "Enabled",
      "Filter": {},
      "AbortIncompleteMultipartUpload": {
        "DaysAfterInitiation": 7
      }
    }
  ]
}

Strategi Dynamic Partitioning Berbasis Waktu (Time-Based Partitioning) #

Agar query engine dapat memindai data secara cepat tanpa harus membaca seluruh isi bucket S3 (yang bisa berukuran petabyte), kita harus membagi file data ke dalam struktur folder partisi yang teratur. Strategi terbaik adalah menggunakan Time-Based Partitioning (Partisi Berbasis Waktu).

Struktur folder ideal yang kompatibel dengan skema partisi Hive adalah:

s3://my-data-lake-bucket/topics/orders/year=2026/month=06/day=08/hour=20/orders_offset_10200.parquet

Mengapa EventTime Jauh Lebih Baik daripada WallClockTime? #

Saat mendistribusikan data ke dalam folder partisi waktu, kita harus memilih metode ekstraksi waktu yang digunakan:

  • WallClockTime: Menggunakan waktu lokal saat worker Connect memproses data tersebut.
    • Masalah: Jika terjadi penundaan pemrosesan konsumen (consumer lag) selama 5 jam, data transaksi yang terjadi pukul 13:00 akan ditulis ke folder partisi pukul 18:00. Hal ini mengacaukan analisis data historis.
  • EventTime: Menggunakan timestamp asli kapan peristiwa tersebut terjadi di dunia nyata (biasanya diekstrak dari kolom payload data seperti transaction_timestamp atau metadata record Kafka).
    • Keuntungan: Menjamin bahwa data selalu mendarat di folder partisi waktu yang benar sesuai dengan waktu kejadian asli, mempermudah query pemulihan data (historical replay).

Parameter Konfigurasi Partisi Waktu #

Berikut adalah parameter yang harus kita konfigurasi untuk menerapkan partisi berbasis EventTime:

# Menggunakan TimeBasedPartitioner bawaan Confluent
partitioner.class=io.confluent.connect.storage.partitioner.TimeBasedPartitioner

# Properti format penulisan direktori di S3
path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH

# Durasi rentang waktu per folder partisi (1 jam = 3600000 ms)
partition.duration.ms=3600000

# Menggunakan extractor waktu EventTime berdasarkan field di dalam payload JSON/Avro
timestamp.extractor=RecordField
timestamp.field=created_at

# Zona waktu yang digunakan untuk standardisasi folder partisi
timezone=UTC
locale=id-ID

Strategi Partisi Lanjutan (Advanced Partitioning Strategies) #

Selain menggunakan partisi berbasis waktu, arsitektur data lake modern sering kali menuntut partisi data berdasarkan kriteria logika bisnis lainnya untuk mendukung isolasi data penyewa (multi-tenancy) atau performa kueri yang lebih tajam.

1. Field Partitioner #

Jika kita ingin memisahkan file data di S3 berdasarkan kolom tertentu (misalnya country_code atau department), kita dapat menggunakan FieldPartitioner.

  • Use Case: Memudahkan kueri lokal untuk hanya menyapu data wilayah tertentu (misal /country=ID/ atau /country=SG/).
  • Properti Konfigurasi: 
    partitioner.class=io.confluent.connect.storage.partitioner.FieldPartitioner
partition.field.name=country_code

2. Hybrid Partitioner (Kombinasi Kustom) #

Sering kali kita ingin menggabungkan pemisahan bisnis dan waktu secara bersamaan, misalnya memisahkan folder berdasarkan ID penyewa (tenant ID), lalu di dalamnya dibagi berdasarkan folder tahun, bulan, dan hari: s3://data-lake/raw-zone/tenant_id=company_a/year=2026/month=06/day=08/file.parquet

Untuk mewujudkan ini, kita dapat merangkai konfigurasi Custom Partitioning dengan menyatukan properti penulisan direktori:

partitioner.class=io.confluent.connect.storage.partitioner.FieldPartitioner
partition.field.name=tenant_id
# Mengaktifkan sub-partisi berbasis waktu di bawah folder field
path.format=tenant_id=${tenant_id}/year=YYYY/month=MM/day=dd

Format Serialisasi dan Kompresi Data #

Format fisik penyimpanan file di S3 sangat memengaruhi performa kueri analisis downstream.

1. Hindari Penggunaan JSON Mentah #

JSON adalah format teks biasa yang sangat boros ruang penyimpanan dan lambat diproses oleh query engine karena setiap baris harus di-parse secara individual.

2. Gunakan Format Kolom (Parquet atau ORC) #

Apache Parquet adalah format penyimpanan berbasis kolom (columnar storage format) yang sangat dioptimalkan untuk analitik data besar.

  • Kompresi Tinggi: Parquet menyimpan data per kolom secara berdekatan, sehingga algoritma kompresi dapat bekerja sangat efisien (karena data dalam satu kolom memiliki tipe yang sama). Ukuran file bisa berkurang hingga 80% dibanding JSON.
  • Column Projection (Pruning): Jika query kita hanya melakukan analisis total penjualan (SELECT SUM(total_price) FROM orders), engine seperti Athena hanya akan mengunduh kolom total_price dari disk S3 dan sepenuhnya mengabaikan kolom lainnya (seperti customer_address atau notes). Hal ini menghemat I/O jaringan hingga 95% dan memangkas biaya Athena secara drastis.

3. Aktifkan Kompresi Tambahan #

Selalu aktifkan kompresi tingkat file. Untuk format Parquet, kompresi bawaan Snappy adalah pilihan standar industri terbaik karena menawarkan keseimbangan yang sangat baik antara rasio kompresi tinggi dan kecepatan dekompresi CPU yang sangat cepat.

Contoh Konfigurasi AWS S3 Sink Connector Komprehensif #

Di bawah ini adalah contoh berkas konfigurasi JSON lengkap untuk mendeploy Amazon S3 Sink Connector yang aman, menggunakan format Parquet dengan kompresi Snappy, partisi berbasis EventTime, serta rotasi file yang optimal untuk menghindari small file problem:

{
  "name": "s3-parquet-sales-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "3",
    "topics": "mysql-db-orders",
    "s3.region": "ap-southeast-1",
    "s3.bucket.name": "badricreativetech-data-lake",
    "topics.dir": "raw-zone",
    
    "//": "--- Konfigurasi Format dan Kompresi Data ---",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "parquet.codec": "snappy",
    
    "//": "--- Pencegahan Small File Problem via Rotasi ---",
    "//": "Menulis ke S3 hanya setelah 100.000 record terkumpul di memori",
    "flush.size": "100000",
    "//": "Atau rotasi file secara berkala setiap 20 menit (1200000 ms)",
    "rotate.interval.ms": "1200000",
    
    "//": "--- Konfigurasi Partisi Waktu Berbasis EventTime ---",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
    "partition.duration.ms": "3600000",
    "timezone": "Asia/Jakarta",
    "locale": "id-ID",
    
    "//": "Mengekstrak waktu partisi dari kolom created_at di payload record",
    "timestamp.extractor": "RecordField",
    "timestamp.field": "created_at",
    
    "//": "--- Converter Avro Terintegrasi Schema Registry ---",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "schema.compatibility": "NONE",
    
    "//": "--- Keamanan dan Akses S3 ---",
    "//": "Menggunakan IAM Role instansi worker Connect (rekomendasi keamanan)",
    "s3.credentials.provider.class": "com.amazonaws.auth.DefaultAWSCredentialsProviderChain"
  }
}

Ringkasan #

  • Cloud Storage Ingestion — Cloud Object Storage (seperti S3 atau GCS) adalah landing zone ideal untuk pengarsipan jangka panjang dan dasar pembangunan ekosistem Data Lake / Lakehouse.
  • Small File Problem — Penulisan file kecil secara berulang membebani performa pembacaan dan melipatgandakan biaya HTTP request. Gunakan parameter flush.size dan rotate.interval.ms untuk mengonsolidasikan file.
  • Time-Based Partitioning — Struktur folder berbasis waktu membatasi jumlah data yang dipindai saat kueri dijalankan (partition pruning), menghemat biaya kueri di AWS Athena.
  • EventTime vs WallClockTime — Selalu gunakan EventTime berbasis kolom transaksi asli untuk memastikan data masuk ke folder partisi waktu yang akurat meskipun terjadi penundaan pemrosesan.
  • Parquet format — Format Parquet biner berbasis kolom dengan kompresi Snappy sangat direkomendasikan karena menghemat ruang disk dan mempercepat kueri analitik secara signifikan.

  ← Sebelumnya: Integrasi Database & CDC   Berikutnya: Integrasi Sistem Eksternal →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact