KAL
Menu

Avro dalam praktik

Skema Avro: record, union, dan nilai default beserta serialisasi klien terhadap Registry.

Daftar isi

Skema Avro adalah dokumen JSON yang menyatakan bentuk data secara lengkap, mulai dari nama record, daftar field, tipe, hingga nilai default. Topik ini melatih penggunaannya secara langsung dengan pustaka fastavro, mulai dari menulis skema record dengan union dan default, melakukan serialisasi menjadi payload biner, hingga membacanya kembali. Bagian akhir menggambarkan alur serializer klien terhadap Schema Registry pada sistem nyata.

Struktur skema record#

Record adalah tipe skema Avro yang paling sering digunakan untuk payload pesan dan setara dengan objek pada JSON. Definisi record menyatakan nama record, daftar field, tipe setiap field, dan nilai default untuk field yang bersifat opsional. Skema berikut mendefinisikan pesan pengukuran sensor dengan dua field opsional.

json Skema record Pengukuran 
{
  "type": "record",
  "name": "Pengukuran",
  "fields": [
    {"name": "sensor", "type": "string"},
    {"name": "suhu", "type": ["null", "double"], "default": null},
    {"name": "kelembapan", "type": ["null", "double"], "default": null}
  ]
}

Tipe suhu dan kelembapan berupa union, yaitu tipe yang menerima lebih dari satu alternatif dan ditulis sebagai daftar. Union ["null", "double"] berarti nilai boleh null maupun bilangan pecahan, sedangkan nilai default null membuat kedua field boleh dihilangkan dari payload. Field sensor tidak memiliki default dan menjadi field wajib, karena payload yang tidak membawanya langsung ditolak pada saat serialisasi.

Record bukan satu-satunya tipe bernama pada Avro, karena tipe lain seperti enum dan fixed juga tersedia untuk kebutuhan khusus. Enum membatasi nilai pada daftar simbol yang tetap, sedangkan fixed menyimpan deretan byte dengan panjang yang pasti. Untuk payload pesan pada Kafka, record adalah tipe yang paling sering digunakan karena meniru bentuk objek pada sebagian besar bahasa pemrograman.

Sebelum digunakan, skema dinormalisasi lewat parse_schema, yaitu proses yang menyusun ulang definisi menjadi bentuk siap pakai dan menautkan nama record untuk resolusi antar versi. Normalisasi dilakukan sekali pada saat aplikasi berjalan, bukan pada setiap pesan. Hasilnya digunakan oleh writer dan reader, yaitu peran yang menulis payload dan peran yang membacanya, dengan skema yang konsisten di kedua sisi.

Serialisasi dan deserialisasi#

Payload Avro disusun tanpa nama field dan tanpa pemisah, sehingga ukurannya jauh lebih kecil daripada JSON. Demo berikut menulis satu record dengan schemaless_writer, yaitu serialisasi payload mentah tanpa header berkas, lalu membacanya kembali dengan schemaless_reader. Pada sistem dengan Registry, payload mentah inilah yang dikirim setelah header lima byte, sedangkan skema diperoleh dari Registry.

python avro-dasar.py 
import io
import json
import fastavro

skema = json.loads('''
{
  "type": "record",
  "name": "Pengukuran",
  "fields": [
    {"name": "sensor", "type": "string"},
    {"name": "suhu", "type": ["null", "double"], "default": null},
    {"name": "kelembapan", "type": ["null", "double"], "default": null}
  ]
}
''')
skema_avro = fastavro.parse_schema(skema)

def kirim(record):
    buf = io.BytesIO()
    fastavro.schemaless_writer(buf, skema_avro, record)
    return buf.getvalue()

def baca(payload):
    return fastavro.schemaless_reader(io.BytesIO(payload), skema_avro)

record = {"sensor": "ruang-1", "suhu": 24.5, "kelembapan": 61.0}
payload = kirim(record)
print("payload Avro      :", payload.hex(" "))
print("jumlah byte Avro  :", len(payload))

json_teks = json.dumps(record)
print("jumlah byte JSON  :", len(json_teks.encode("utf-8")))

print("terbaca:", baca(payload))

# union: suhu boleh None (cabang null dari ["null", "double"])
record_tanpa_suhu = {"sensor": "ruang-1", "suhu": None, "kelembapan": 55.0}
payload2 = kirim(record_tanpa_suhu)
print("terbaca:", baca(payload2))
text Keluaran 
payload Avro      : 0e 72 75 61 6e 67 2d 31 02 00 00 00 00 00 80 38 40 02 00 00 00 00 00 80 4e 40
jumlah byte Avro  : 26
jumlah byte JSON  : 55
terbaca: {'sensor': 'ruang-1', 'suhu': 24.5, 'kelembapan': 61.0}
terbaca: {'sensor': 'ruang-1', 'suhu': None, 'kelembapan': 55.0}

Payload sepanjang 26 byte menyimpan tiga bagian. Byte 0e adalah panjang string 7 dalam varint zigzag, diikuti UTF-8 dari ruang-1. Byte 02 menandai cabang union untuk double pada field suhu, diikuti delapan byte bilangan pecahan, lalu pola yang sama berulang untuk kelembapan. Perbandingan dengan JSON berukuran 55 byte memperlihatkan sumber pemangkasan utamanya, yaitu nama field yang tidak pernah ikut dalam payload. Record kedua memperlihatkan bahwa union bekerja pada cabang null tanpa masalah.

Avro juga menyediakan format berkas yang menyisipkan skema lengkap pada header berkasnya, sehingga satu berkas dapat memuat ribuan record tanpa Registry, dan header itu diawali empat byte Obj\x01. Format semacam itu berguna untuk arsip dan pertukaran berkas, tetapi tidak cocok untuk payload Kafka, karena setiap pesan akan menanggung ulang skema yang sebenarnya sudah tersimpan di Registry. Fungsi schemaless_writer pada demo menghindari header itu dan menghasilkan payload mentah yang menjadi isi pesan.

Nilai default dan field wajib#

Nilai default diisi oleh writer pada saat serialisasi, sehingga payload yang dikirim tetap lengkap. Demo berikut menulis record yang hanya membawa field sensor, kemudian membaca kembali hasilnya. Bagian kedua menulis record yang kehilangan field wajib untuk memperlihatkan penolakan dini di sisi producer.

python avro-default.py 
import io
import json
import fastavro

skema = json.loads('''
{
  "type": "record",
  "name": "Pengukuran",
  "fields": [
    {"name": "sensor", "type": "string"},
    {"name": "suhu", "type": ["null", "double"], "default": null},
    {"name": "kelembapan", "type": ["null", "double"], "default": null}
  ]
}
''')
skema_avro = fastavro.parse_schema(skema)

# suhu dan kelembapan tidak dikirim, nilai default null mengisi keduanya
buf = io.BytesIO()
fastavro.schemaless_writer(buf, skema_avro, {"sensor": "ruang-2"})
data = buf.getvalue()
print("terbaca:", fastavro.schemaless_reader(io.BytesIO(data), skema_avro))

# sensor wajib ada dan tidak punya default
try:
    fastavro.schemaless_writer(io.BytesIO(), skema_avro, {"suhu": 20.0})
except ValueError as e:
    print(f"gagal: {e}")
text Keluaran 
terbaca: {'sensor': 'ruang-2', 'suhu': None, 'kelembapan': None}
gagal: no value and no default for sensor

Record pertama terbaca lengkap dengan suhu dan kelembapan bernilai None, meskipun keduanya tidak pernah dikirim. Field wajib yang hilang langsung gagal dengan pesan no value and no default for sensor, sehingga payload cacat tidak pernah mencapai jaringan. Kegagalan dini seperti ini (fail fast) adalah salah satu alasan serialisasi berbasis skema lebih aman daripada JSON bebas.

Spesifikasi Avro menetapkan bahwa default pada field bertipe union harus cocok dengan cabang pertamanya, dan skema demo mematuhi ketentuan itu dengan null untuk union ["null", "double"]. fastavro tidak menolak skema yang melanggar ketentuan tersebut pada saat parse_schema dijalankan, sehingga penjagaannya menjadi tanggung jawab penulis skema. Penyandian union sendiri hanya menyimpan indeks cabang tanpa nama tipe, sehingga nilai null pada field opsional hanya menempati satu byte indeks.

Klien Avro dengan Registry#

Producer nyata tidak menyusun header dan payload satu per satu, melainkan menyerahkannya kepada serializer klien. Serializer memvalidasi record terhadap skema, meminta ID skema dari Registry, lalu menanamkan header lima byte di depan payload. Sisi consumer bekerja simetris melalui deserializer yang membaca ID pada header dan mengambil skema dari Registry sebelum menafsirkan payload. Potongan berikut menggunakan pustaka confluent-kafka untuk memperlihatkan alur tersebut.

python Alur serializer klien terhadap Registry 
# tidak-dijalankan
# memerlukan paket confluent-kafka dan Schema Registry yang berjalan di localhost:8081
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer

klien_registry = SchemaRegistryClient({"url": "http://localhost:8081"})
serializer = AvroSerializer(klien_registry, skema_avro)

# serializer mengembalikan bytes berformat kawat: magic 0 + ID skema + payload
bytes_pesan = serializer(record)

Deserializer menerima byte yang sama, memisahkan header, mengambil skema writer berdasarkan ID, lalu menerapkan resolusi dengan skema reader milik consumer. Resolusi antar skema inilah yang menentukan apakah payload lama masih bisa dibaca oleh consumer baru. Pemisahan peran ini membuat klien tetap ringkas, karena header dan pemilihan skema tidak pernah ditangani kode pesan secara manual.

Serializer memilih subject menggunakan strategi nama yang sama seperti pada pendaftaran skema, sehingga skema yang dikirim dan skema yang dibaca selalu berkumpul pada subject yang benar. Permintaan skema menyimpan hasilnya di cache lokal, sehingga pemanggilan ke Registry berhenti setelah skema pertama setiap ID diterima. Dengan cache tersebut, biaya serialisasi per pesan turun menjadi pekerjaan biner murni tanpa jaringan.

Ringkasan#

  • Skema Avro menyatakan record lengkap dengan tipe, union, dan default, kemudian dinormalisasi sekali lewat parse_schema.
  • Payload Avro memangkas nama field sehingga lebih kecil daripada JSON, dan nilai default diisi pada saat serialisasi.
  • Field wajib yang hilang gagal lebih awal di sisi producer, sedangkan serializer Registry menambahkan header ID skema di depan payload.

Referensi#