Bekerja dengan berkas

Topik ini membahas penanganan berkas pada pekerjaan data engineering dengan modul pathlib, mulai dari menjelajah direktori sampai membaca berkas besar secara aman. Satu berkas CSV penjualan dipakai sebagai bahan kerja, kemudian diringkas dengan pola streaming, yaitu pemrosesan data secara bertahap baris demi baris tanpa menunggu seluruh berkas dimuat. Setelah topik ini, navigasi berkas dan pembacaan berkas raksasa dapat dilakukan dengan pola yang baku dan hemat memori.

Menjelajah direktori dengan pathlib#

Modul pathlib menyerahkan seluruh pekerjaan jalur berkas kepada objek Path, sehingga navigasi direktori tidak lagi bergantung pada perangkaian string yang rentan salah. Objek Path mewakili sebuah jalur berkas atau direktori beserta operasinya, mulai dari pemeriksaan keberadaan sampai pencarian pola. Operator / menyusun subdirektori secara lintas sistem operasi, sehingga kode yang sama berjalan pada Linux maupun Windows. Atribut name memberi nama berkas tanpa jalurnya, sedangkan metode glob mencari berkas yang cocok dengan pola tertentu.

Contoh berikut menjalankan penjelajahan kecil pada direktori kerja yang memuat berkas penjualan.csv.

from pathlib import Path

kini = Path.cwd()
print(kini.name)
print(sorted(p.name for p in kini.glob("*.csv")))
print(Path("penjualan.csv").exists())
print(Path("data") / "penjualan.csv")

bekerja-dengan-berkas
['penjualan.csv']
True
data/penjualan.csv

Baris pertama keluaran menunjukkan nama direktori kerja, baris kedua menampilkan hasil pencarian glob, dan baris ketiga mengonfirmasi keberadaan berkas CSV. Baris keempat memperlihatkan operator / yang menghasilkan jalur baru tanpa menyentuh sistem berkas, karena Path hanya mewakili jalur sampai operasi baca atau tulis benar-benar dilakukan. Metode lain yang sering dipakai meliputi suffix untuk ekstensi berkas, mkdir untuk membuat direktori, dan stat untuk ukuran berkas, sehingga sebagian besar skrip tidak lagi memerlukan modul os.path. Untuk berkas kecil, Path.read_text() membaca seluruh isi sekaligus, tetapi cara itu memiliki batasan yang dibahas di bagian streaming.

Membaca CSV baris demi baris#

Pembacaan CSV yang aman dilakukan dengan modul csv yang bekerja di atas aliran baris, bukan dengan memuat seluruh isi berkas ke dalam memori. Kelas csv.DictReader membaca baris pertama sebagai judul kolom, lalu mengubah setiap baris berikutnya menjadi kamus yang kuncinya adalah nama kolom. Fungsi dengan pernyataan yield disebut generator, yaitu fungsi yang menghasilkan nilai satu per satu hanya saat diminta, sehingga baris yang belum dibutuhkan tidak pernah menempati memori. Pernyataan with menutup berkas secara otomatis setelah perulangan selesai, termasuk ketika terjadi error di tengah proses.

Berkas CSV yang dipakai sepanjang topik ini berisi lima baris data di bawah satu baris judul kolom.

produk,jumlah,harga_satuan
buku tulis,3,4500
pensil,10,2000
buku tulis,2,4500
penghapus,0,1500
spidol,4,3500

Berkas berikut merangkai semuanya menjadi skrip ringkasan yang dapat langsung dijalankan dari direktori tempat penjualan.csv berada.

"""Meringkas penjualan dari berkas CSV secara streaming baris demi baris."""
import csv
from collections.abc import Iterator
from pathlib import Path


def baris_penjualan(berkas: Path) -> Iterator[dict[str, str]]:
    """Menghasilkan satu baris CSV pada satu waktu, tanpa memuat seluruh berkas."""
    with berkas.open(newline="", encoding="utf-8") as sumber:
        pembaca = csv.DictReader(sumber)
        for baris in pembaca:
            yield baris


def total_per_produk(berkas: Path) -> dict[str, float]:
    """Mengagregasi total penjualan per produk dari aliran baris."""
    hasil: dict[str, float] = {}
    jumlah_baris = 0
    for baris in baris_penjualan(berkas):
        jumlah_baris += 1
        produk = baris["produk"]
        total = int(baris["jumlah"]) * float(baris["harga_satuan"])
        hasil[produk] = hasil.get(produk, 0.0) + total
    print(f"baris data dibaca: {jumlah_baris}")
    return hasil


if __name__ == "__main__":
    berkas = Path("penjualan.csv")
    if not berkas.exists():
        raise SystemExit(f"berkas {berkas} tidak ditemukan di direktori kerja")
    print(f"berkas: {berkas.name}")
    for produk, total in total_per_produk(berkas).items():
        print(f"{produk}: {total:.0f}")

berkas: penjualan.csv
baris data dibaca: 5
buku tulis: 22500
pensil: 20000
penghapus: 0
spidol: 14000

Keluaran menunjukkan lima baris data terbaca dan total per produk teragregasi dalam satu kali lewatan. Fungsi baris_penjualan menghasilkan satu kamus per baris, sedangkan fungsi total_per_produk hanya menyimpan agregasinya. Memori yang dipakai bertambah seiring banyaknya produk yang berbeda, bukan seiring banyaknya baris, dan itulah sifat yang diinginkan pada berkas besar.

Streaming untuk berkas besar#

Pola streaming menjaga penggunaan memori tetap konstan karena hanya baris yang sedang diproses yang berada di memori. Perbandingannya mudah diukur dari cara berkas dibuka. Pemanggilan read_text() atau read() memuat seluruh isi berkas sebagai satu objek tunggal, sehingga berkas 10 GB pada mesin dengan RAM 8 GB akan gagal diproses. Iterasi atas objek berkas atau atas generator seperti baris_penjualan hanya menyimpan satu baris pada satu waktu, sehingga berkas 10 GB dapat diproses pada mesin yang jauh lebih kecil asalkan prosesnya selesai dalam waktu yang wajar.

Sifat ini penting karena berkas ekstraksi di pekerjaan data engineering sering tumbuh tanpa batas yang terduga pada awalnya. Skrip yang dirancang streaming sejak hari pertama tidak perlu ditulis ulang ketika berkasnya berpindah dari seratus baris ke seratus juta baris. Pola umumnya tetap sama, yaitu buka berkas, proses baris demi baris, simpan agregasi atau tulis keluaran, lalu tutup berkas.

Penggunaan memori pada pola ini juga dapat diperhitungkan sejak awal. Kamus agregasi tumbuh sebanding dengan banyaknya kunci yang berbeda, sedangkan aliran barisnya memakan ruang yang tetap. Apabila agregasinya pun melebihi memori, hasil antara ditulis ke berkas kedua per kelompok, atau pekerjaan dipindahkan ke alat yang dirancang untuk data besar. Keputusan semacam itu dibuat karena ukuran datanya, bukan karena pola kodenya perlu berubah.

Kesalahan umum#

Penanganan berkas memiliki beberapa jebakan yang muncul berulang kali pada kode pipeline, dan semuanya jarang terlihat saat data uji masih kecil. Tiga yang paling sering terjadi beserta perbaikannya adalah sebagai berikut.

• Memanggil open() untuk CSV tanpa argumen newline="", sehingga muncul baris kosong tambahan ketika kode dijalankan pada Windows. Perbaikannya, selalu sertakan newline="" saat membuka berkas untuk modul csv.
• Memakai read_text() atau read() pada berkas yang ukurannya tidak terkendali, sehingga memori habis justru ketika berkas nyata tiba di pipeline. Perbaikannya, proses berkas lewat iterasi atau generator sebagaimana hitung-penjualan.py.
• Merangkai jalur dengan penjumlahan string seperti "data/" + nama, sehingga pemisah direktori keliru ketika kode berpindah sistem operasi. Perbaikannya, gunakan objek Path dan operator / untuk setiap penyusunan jalur.

Ringkasan#

• Objek Path dari pathlib menyatukan navigasi, pemeriksaan, dan pencarian berkas tanpa perangkaian string manual.
• csv.DictReader mengubah setiap baris menjadi kamus berdasarkan judul kolom pada baris pertama.
• Fungsi generator menghasilkan baris satu per satu sehingga penggunaan memori konstan berapa pun ukuran berkasnya.
• Pola streaming yang dibangun sejak awal membuat skrip tetap layak dipakai ketika berkas tumbuh jauh lebih besar.

Referensi#

• pathlib — Dokumentasi resmi Python — Rujukan objek Path, metode jelajah direktori, dan operator penyusunan jalur.
• csv — Dokumentasi resmi Python — Rujukan csv.reader dan csv.DictReader, termasuk aturan argumen newline="".