KAL
Menu

Strategi baca/tulis

Evolusi schema, mergeSchema, dan mode penulisan.

Daftar isi

Strategi baca dan tulis menyangkut keputusan yang berlaku jangka panjang pada data yang disimpan, yaitu bagaimana mode penulisan menghadapi keluaran yang sudah ada dan bagaimana pembacaan menghadapi berkas yang schemanya tidak seragam. Topik ini membahas empat mode penulisan beserta perilaku bawaannya, evolusi schema pada penambahan kolom, dan opsi mergeSchema sebagai jawabannya. Setelah membacanya, penulisan berulang atas folder yang sama dan perubahan struktur data dari waktu ke waktu dapat dikelola tanpa kejutan.

Empat mode penulisan#

Penulis DataFrame menyediakan empat mode, yaitu append untuk menambah di belakang keluaran yang ada, overwrite untuk menggantinya, errorifexists sebagai bawaan yang gagal bila target sudah ada, dan ignore yang membiarkan target lama tetap utuh tanpa menulis apa pun. Bawaan yang gagal itu dirancang sebagai pengaman, karena penulisan yang salah target tidak layak menghapus data secara diam-diam.

Pilihan mode mengikuti maksud penulisan. append cocok untuk data yang bertambah dari waktu ke waktu, misalnya catatan harian ke tabel yang sama. overwrite cocok untuk keluaran yang dibangun ulang penuh dari sumber, misalnya tabel ringkasan yang dihitung ulang setiap proses. Dua mode sisanya memainkan peran pengaman pada skrip yang dijalankan ulang, dan memakai bawaan secara sadar lebih baik daripada menambahkan overwrite di mana-mana karena dianggap praktis.

Skrip berikut memperagakan append dua kali diikuti overwrite pada folder yang sama, dan perubahan jumlah barisnya menunjukkan efek setiap mode. Folder target sengaja dihapus di awal skrip agar hasilnya selalu sama ketika skrip diulang, dan penghapusan itu bukan bagian dari mode penulisan, melainkan penataan bahan peraga.

python mode_tulis.py 
import shutil

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("mode-tulis")
    .config("spark.sql.shuffle.partitions", "2")
    .getOrCreate()
)

df = spark.createDataFrame(
    [(1, "Ali", 85), (2, "Budi", 90)],
    "id long, nama string, nilai long",
)
target = "/tmp/lokasi/pq_mode"
shutil.rmtree(target, ignore_errors=True)

df.write.mode("append").parquet(target)
df.write.mode("append").parquet(target)
print("setelah dua append:", spark.read.parquet(target).count())

df.write.mode("overwrite").parquet(target)
print("setelah overwrite:", spark.read.parquet(target).count())

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
setelah dua append: 4
setelah overwrite: 2

Dua append menggandakan jumlah baris menjadi 4 karena setiap penulisan menambah berkas partisi baru, dan overwrite mengembalikannya ke 2 karena seluruh isi folder diganti dengan keluaran yang baru. Perlu dicatat bahwa overwrite menghapus berkas lama sebelum menulis yang baru, sehingga kegagalan di tengah proses berarti keluaran lama sudah tidak ada. Untuk penggantian yang aman pada data produksi, format tabel seperti Delta Lake yang dibahas pada modul ekosistem menyediakan transaksi di atas operasi yang sama.

Evolusi schema dan mergeSchema#

Data yang terus bertambah hampir pasti berubah strukturnya. Kolom baru muncul pada batch berikutnya, dan berkas lama di folder yang sama tetap berbentuk lama. Parquet menyimpan schema di footernya, sehingga pembacaan folder yang isinya tidak seragam menuntut keputusan, yaitu memakai schema salah satu berkas atau menggabungkan semuanya.

Tanpa opsi tambahan, pembaca Parquet memilih schema dari salah satu berkas di folder, dan pilihan itu tidak dapat diandalkan ketika berkas-berkasnya berbeda, karena kolom yang hanya ada di sebagian berkas dapat hilang dari pembacaan secara diam-diam. Opsi mergeSchema memerintahkan penggabungan schema seluruh berkas, sehingga kolom yang ada di berkas mana pun tetap muncul, dan baris dari berkas yang tidak memuatnya dibaca sebagai null.

Skrip berikut menulis tabel dasar tanpa kolom kelas, lalu menambahkan berkas baru yang memuat kolom itu, dan membaca keduanya dengan mergeSchema.

python merge_schema.py 
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("merge-schema")
    .config("spark.sql.shuffle.partitions", "2")
    .getOrCreate()
)

dasar = spark.createDataFrame([(1, "Ali")], "id long, nama string")
target = "/tmp/lokasi/pq_merge"
dasar.write.mode("overwrite").parquet(target)

tambahan = spark.createDataFrame([(2, "Budi", "10A")], "id long, nama string, kelas string")
tambahan.write.mode("append").parquet(target)

print("=== baca dengan mergeSchema ===")
spark.read.option("mergeSchema", "true").parquet(target).printSchema()

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== baca dengan mergeSchema ===
root
 |-- id: long (nullable = true)
 |-- nama: string (nullable = true)
 |-- kelas: string (nullable = true)

Schema hasil gabungan memuat kelas meskipun berkas pertama tidak memuatnya, dan itulah kontrak yang diberikan mergeSchema. Opsi ini dipanggil pada pembaca sebagaimana contoh, dan juga dapat diatur sebagai kunci konfigurasi agar berlaku untuk seluruh sesi pembacaan tabel yang memang dirancang mengizinkan evolusi.

Batas evolusi schema#

mergeSchema menggabungkan kolom dengan aman ketika perbedaannya berupa kolom yang ditambahkan atau tidak ada, tetapi perbedaan tipe pada kolom bernama sama tidak diselesaikan dengan begitu saja. Kolom yang bertipe int pada satu berkas dan long pada berkas lain dipromosikan menurut aturan tipe yang lebih luas bila memungkinkan, sedangkan perbedaan yang tidak dapat disetujui memunculkan error saat pembacaan. Karena itu evolusi schema tetap memerlukan disiplin, yaitu perubahan tipe diumumkan dan dilakukan secara sadar, bukan dibiarkan pada isi berkas.

Pada folder terpartisi, kolom partisi ikut bergabung ke schema secara otomatis dari nama folder, dan perpaduan schema datanya berjalan di atas konvensi itu. Pembacaan folder yang bertahun-tahun berevolusi karenanya tetap dapat dilakukan, selama penambahan kolomnya bersifat murni penambahan dan bukan perubahan makna kolom lama, yang justru paling sulit dideteksi oleh alat mana pun.

Kesalahan umum#

Kesalahan pertama adalah memakai overwrite pada target yang dipakai bersama tanpa menyadari seluruh isinya diganti. Tabel yang dibaca pihak lain mendadak berubah atau kosong, dan penyebabnya tidak tercatat di mana pun. Pisahkan folder keluaran sementara dari folder tabel yang dikonsumsi, atau pakai format bertransaksi untuk penggantian yang aman.

Kesalahan kedua adalah membaca folder yang schemanya berevolusi tanpa mergeSchema, lalu merujuk kolom yang baru. Pembacaan berjalan tanpa kolom itu dan transformasi berikutnya gagal dengan error kolom tidak dikenal, atau lebih buruk, berjalan dengan schema yang kehilangan kolom baru itu secara diam-diam. Aktifkan opsi tersebut pada folder yang memang dirancang berubah.

Kesalahan ketiga adalah mengubah tipe kolom pada penulisan baru tanpa memeriksa kompatibilitasnya. Pembacaan dengan mergeSchema yang tadinya berhasil menjadi gagal karena dua tipe tidak dapat disetujui, dan perbaikannya memerlukan penulisan ulang berkas lama. Perubahan tipe dikelola sebagai keputusan schema, bukan efek samping skrip.

Ringkasan#

  • Empat mode penulisan adalah append, overwrite, errorifexists sebagai bawaan, dan ignore, dan pilihannya mengikuti maksud penulisan atas keluaran yang sudah ada.
  • Berkas Parquet membawa schema di footernya, sehingga folder yang isinya tidak seragam menuntut keputusan pembacaan.
  • mergeSchema menggabungkan schema seluruh berkas sehingga kolom tambahan tetap muncul, dengan baris lama terbaca sebagai null pada kolom barunya.
  • Evolusi tipe kolom tidak otomatis, dan perubahan tipe tetap dikelola sebagai keputusan sadar agar pembacaan tidak gagal.

Referensi#