KAL
Menu

Membaca data

Pembaca CSV, JSON, dan Parquet beserta opsi pentingnya.

Daftar isi

Pembacaan data adalah langkah pertama hampir setiap pekerjaan Spark, dan format berkas menentukan perilaku pembacanya. Topik ini membahas tiga format yang paling umum, yaitu CSV, JSON Lines, dan Parquet, beserta opsi-opsi pentingnya melalui API spark.read. Setelah membacanya, ketiga format tersebut dapat dibaca dengan nama kolom yang benar, dan perbedaan perilaku schema di antara ketiganya dapat dijelaskan.

Data contoh di dalam topik#

Dua berkas data kecil disertakan sebagai aset topik ini, yaitu daftar nilai ujian dalam format CSV dan catatan sensor suhu dalam format JSON Lines. Skrip pada topik ini membaca keduanya melalui path relatif, karena skrip dijalankan dari direktori topik tempat kedua berkas itu berada.

csv nilai_ujian.csv 
id,nama,kelas,nilai
1,Ali,10A,85
2,Budi,10A,90
3,Citra,10B,78
4,Dewi,10B,92
5,Eko,10A,88
6,Farah,10B,95
json sensor_suhu.json 
{"sensor": "S1", "lokasi": "ruang-server", "suhu": 24.5}
{"sensor": "S2", "lokasi": "ruang-lab", "suhu": 23.8}
{"sensor": "S3", "lokasi": "perpustakaan", "suhu": 25.1}
{"sensor": "S4", "lokasi": "ruang-guru", "suhu": 24.0}

Berkas CSV memuat baris header berisi id,nama,kelas,nilai diikuti enam baris data. Berkas JSON Lines memuat empat baris, dan setiap baris adalah satu objek JSON utuh dengan kunci sensor, lokasi, serta suhu. Format satu objek per baris inilah yang disebut JSON Lines dan merupakan format yang dibaca pembaca JSON bawaan Spark, bukan satu berkas JSON tunggal berisi satu objek.

Membaca CSV dan arti opsi header#

Skrip berikut membaca kedua berkas aset dengan path relatif. Bagian awal skrip sengaja membaca CSV dua kali, sekali tanpa opsi apa pun dan sekali dengan .option("header", "true"), agar perbedaannya terlihat langsung pada schema dan isi.

python baca_csv_json.py 
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("membaca-csv-json")
    .getOrCreate()
)

print("=== csv tanpa opsi header ===")
tanpa_header = spark.read.csv("nilai_ujian.csv")
tanpa_header.printSchema()
tanpa_header.orderBy("_c0").show()

print("=== csv dengan opsi header true ===")
dengan_header = spark.read.option("header", "true").csv("nilai_ujian.csv")
dengan_header.printSchema()
dengan_header.orderBy("id").show()

print("=== json lines ===")
suhu = spark.read.json("sensor_suhu.json")
suhu.printSchema()
suhu.orderBy("sensor").show()

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== csv tanpa opsi header ===
root
 |-- _c0: string (nullable = true)
 |-- _c1: string (nullable = true)
 |-- _c2: string (nullable = true)
 |-- _c3: string (nullable = true)

+---+-----+-----+-----+
|_c0|  _c1|  _c2|  _c3|
+---+-----+-----+-----+
|  1|  Ali|  10A|   85|
|  2| Budi|  10A|   90|
|  3|Citra|  10B|   78|
|  4| Dewi|  10B|   92|
|  5|  Eko|  10A|   88|
|  6|Farah|  10B|   95|
| id| nama|kelas|nilai|
+---+-----+-----+-----+

=== csv dengan opsi header true ===
root
 |-- id: string (nullable = true)
 |-- nama: string (nullable = true)
 |-- kelas: string (nullable = true)
 |-- nilai: string (nullable = true)

+---+-----+-----+-----+
| id| nama|kelas|nilai|
+---+-----+-----+-----+
|  1|  Ali|  10A|   85|
|  2| Budi|  10A|   90|
|  3|Citra|  10B|   78|
|  4| Dewi|  10B|   92|
|  5|  Eko|  10A|   88|
|  6|Farah|  10B|   95|
+---+-----+-----+-----+

=== json lines ===
root
 |-- lokasi: string (nullable = true)
 |-- sensor: string (nullable = true)
 |-- suhu: double (nullable = true)

+------------+------+----+
|      lokasi|sensor|suhu|
+------------+------+----+
|ruang-server|    S1|24.5|
|   ruang-lab|    S2|23.8|
|perpustakaan|    S3|25.1|
|  ruang-guru|    S4|24.0|
+------------+------+----+

Tanpa opsi header, pembaca CSV menamai kolom secara otomatis menjadi _c0 sampai _c3, dan baris id,nama,kelas,nilai masuk sebagai baris data biasa yang muncul paling bawah pada tabel. Dengan .option("header", "true"), baris pertama dipakai sebagai nama kolom dan tidak lagi menjadi data. Perhatikan juga bahwa seluruh kolom CSV bertipe string pada kedua pembacaan. Pembaca CSV tanpa inferSchema tidak menebak tipe apa pun dan menganggap setiap nilai sebagai teks, sehingga kolom angka tetap terbaca sebagai teks.

Setiap pembacaan diakhiri printSchema() dan show() yang didahului orderBy dengan kunci unik, agar struktur dan urutan keluarannya tetap sama pada setiap eksekusi.

Membaca JSON Lines#

Pembacaan JSON berperilaku berbeda dari CSV. Setiap objek JSON membawa pasangan nama dan nilai, sehingga nama kolom datang dari data itu sendiri dan opsi header tidak diperlukan. Pembaca JSON juga selalu menyimpulkan schema dengan mengambil sampel isi berkas, tanpa opsi tambahan apa pun. Pada keluaran skrip di atas, kolom suhu yang berisi angka desimal terbaca sebagai double, sedangkan sensor dan lokasi terbaca sebagai string.

Kolom hasil inferensi JSON tercetak terurut alfabetis pada keluaran, sehingga urutannya tidak mengikuti urutan kemunculan kunci di berkas. Kolom mana pun dapat dipilih sebagai kunci orderBy, dan skrip memakai sensor karena nilainya unik. Apabila salah satu baris JSON rusak, mode default bernama PERMISSIVE tidak menghentikan pembacaan, tetapi mengisi kolom yang gagal dibaca dengan nilai kosong, sehingga data tetap masuk dan cacatnya perlu diperiksa kemudian.

Parquet: schema ikut tersimpan#

Parquet menyimpan data per kolom beserta metadata schema di dalam berkasnya sendiri. Penulisannya diperagakan oleh dua skrip yang dijalankan berurutan. Skrip pertama membuat sebuah DataFrame kecil, mencetak schema sebelum ditulis, lalu menuliskannya sebagai folder Parquet di /tmp/lokasi/parquet_ujian.

python tulis_parquet.py 
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("menulis-parquet")
    .getOrCreate()
)

df = spark.createDataFrame(
    [(1, "Ali", 85.5), (2, "Budi", 90.0), (3, "Citra", 78.25)],
    "id long, nama string, nilai double",
)

print("=== schema sebelum ditulis ===")
df.printSchema()

df.write.mode("overwrite").parquet("/tmp/lokasi/parquet_ujian")
print("data ditulis ke /tmp/lokasi/parquet_ujian")

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== schema sebelum ditulis ===
root
 |-- id: long (nullable = true)
 |-- nama: string (nullable = true)
 |-- nilai: double (nullable = true)

data ditulis ke /tmp/lokasi/parquet_ujian

Skrip kedua membaca kembali folder tersebut untuk menunjukkan bahwa tipe data ikut terbawa tanpa proses inferensi apa pun.

python baca_parquet.py 
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("membaca-parquet")
    .getOrCreate()
)

parquet_df = spark.read.parquet("/tmp/lokasi/parquet_ujian")

print("=== schema setelah dibaca kembali ===")
parquet_df.printSchema()
parquet_df.orderBy("id").show()

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== schema setelah dibaca kembali ===
root
 |-- id: long (nullable = true)
 |-- nama: string (nullable = true)
 |-- nilai: double (nullable = true)

+---+-----+-----+
| id| nama|nilai|
+---+-----+-----+
|  1|  Ali| 85.5|
|  2| Budi| 90.0|
|  3|Citra|78.25|
+---+-----+-----+

Schema sebelum ditulis dan sesudah dibaca kembali identik, termasuk perbedaan antara long pada kolom id dan double pada kolom nilai. Tidak ada inferensi yang berjalan, karena tipe data tersimpan dalam metadata Parquet dan pembaca tinggal mematuhinya. Hasil penulisan berupa folder berisi satu atau lebih berkas part-*.parquet serta berkas penanda _SUCCESS, dan spark.read.parquet() menerima path folder tersebut sebagai satu kesatuan data.

Folder keluaran sengaja diletakkan di /tmp dan bukan di samping skrip, karena hasil eksekusi bukan bagian dari data sumber dan tidak boleh mengotori direktori topik.

Kesalahan umum#

Kesalahan pertama adalah lupa .option("header", "true") pada CSV yang memiliki baris header. Akibatnya, kolom bernama _c0 sampai _cN dan baris header menjadi data, sehingga agregasi yang merujuk nama kolom gagal atau baris header ikut terhitung sebagai data. Perbaikannya cukup menambahkan opsi header, lalu periksa ulang dengan printSchema().

Kesalahan kedua adalah menganggap kolom CSV berisi angka sudah bertipe angka. Tanpa inferSchema atau schema eksplisit, seluruh kolom CSV bertipe string, sehingga pengurutan dilakukan secara leksikografis dan nilai 9 terurut setelah 10. Perbaikannya adalah menyatakan tipe secara eksplisit atau mengaktifkan inferensi, sesuai bahasan pada topik schema.

Kesalahan ketiga adalah menulis hasil pemrosesan ke folder data sumber atau ke direktori topik. Penulisan Spark menghasilkan folder berkas partisi yang akan menumpuk berkas sumber bila ditulis ke path yang sama. Pisahkan folder keluaran, misalnya di bawah /tmp untuk eksperimen lokal, dan jaga direktori topik tetap berisi skrip serta data sumber saja.

Ringkasan#

  • spark.read.csv(), spark.read.json(), dan spark.read.parquet() menghasilkan DataFrame dari tiga format yang paling umum, dengan opsi yang dirangkai melalui .option().
  • Tanpa opsi header, pembaca CSV menamai kolom _c0 dan seterusnya serta menjadikan baris header data, dan tanpa inferensi seluruh kolom CSV bertipe string.
  • Pembaca JSON Lines selalu menyimpulkan schema dari sampel berkas, sedangkan Parquet membawa schema tersimpan sehingga tipe data kembali persis tanpa tebakan.
  • Hasil penulisan Spark berupa folder partisi dan sebaiknya diarahkan ke folder keluaran seperti /tmp, bukan ke folder data sumber.

Referensi#