Bikin ETL Sederhana Menggunakan Python dan PostgreSQL

Cerita nya lagi gabut dan bikin thumbnail wkwkwk

BJIR GG COK awkkawkakw

Lanjut ah, jadi hari ini gw cuman belajar singkat buat masukin data berformat csv ke database PostgreSQL pakai Pandas sebagai ETL tools nya. Gw keinget dulu tuh pas pernah interview sebagai data engineer magang, ditanya pernah bikin ETL scripting gak? Pernah lah gw bilang, sampai tools nya gw sebut pakai Airflow, bikin DAG scheduler segala macam, pindahin ke data lake di HDFS yang gw setup sendiri. Tapi rupanya pertanyaan dia gak sampai situ, sekedar konek ke database doang. Di saat itu gw emang sama sekali gak pernah hands-on, cuma sekedar tau konsep, cara kerja, atau library yg dipakai doang, jadi jujurlah kan, gw rasa karena core nya memang di situ keperluannya dan gw juga kurang di situ jadi gak terpilih anjas wkwkwk.

Opening macam apa itu.

Yak mungkin itu cuma intermezzo doang buat nginget bahwa gw pernah kepikiran buat ngerjain ni proyek kecil-kecilan, harusnya sih kecil ya, karena emang cuman pindahin data ke dalam database doang. Jadi adapun tools yang dipakai, yaitu:

• Python
• Library psycopg2, csv
• PostgreSQL

Jadi awalnya tuh gw kepikiran juga, gimana caranya gw mau ngebuild skill gw khususnya di bidang data engineer, takutnya gw mikir kejauhan kayak langsung pakai tools Airflow, bikin data warehouse yang ada CDC nya, belajar dbt, bikin streaming pipeline, processing pakai Spark, dan lainnya. Sejujurnya itu yg bikin gw mikir kejauhan, kayak mending gw urutin dulu aja gw kerjain semuanya tapi gw coba rangkai dulu dari yang sekiranya paling mudah, fundamental, dan membangun fondasi gw.

Ketemulah jawabannya, gw berencana untuk coba pindahin data ke database, trus ntah mungkin gw lakukan etl lagi ke database lain tapi data nya udah gw lakukan pre-processing (bersihin data, struktur nya, kualitas nilai nya, dan sebagainya), kemudian bisa juga gw pindahin ke data warehouse dengan catatan skema nya nanti tentu saja mengikuti best practice dengan mengikuti pemahaman penggunaan skema kayak star schema ataupun dimensional schema (karena jujur gw masih belum pernah hands-on bikin skema data warehouse). Gak sampai situ, bisa data nya gw analisis dan bikin dashboard, atau gw dump kemudian gw export ke data lake. Nah kan kurang lebih mencakup banyak aspek tuh, jadi ya gak apa dulu semuanya gw buiild dari 0.

Balik lagi, sekarang gw akan menjelaskan proyek kecil-kecilan gw ini

Di sini kita perlu import library untuk bisa konek ke database nya, di sini gw pakai pyscopg2 untuk koneksi ke database PostgreSQL

Berikutnya di sini gw coba bikin koneksi pakai object connect() dari library psycopg2, parameter nya ikutin aja. Oh iya di setiap cell ini juga gw buat try and exception biar setiap gw melakukan eksekusi ke database bisa tau hasilnya, kalau berhasil ya diem klo gagal ya tampilkan error nya apa.

Next gw bikin cursor, cursor ini kek objek atau mesin yang nantinya akan membantu kita melakukan query ke dalam database melalui Python code.

Berikutnya ada set_session(autocommit=True), sedikit yang gw pahami, jadi di psycopg2 ini cara agar setiap query ataupun transaksi yang dikirimkan ke database postgre itu gak langsung berdampak di database, sehingga pada dasarnya perlu dilakukan commit() terlebih dahulu agar query yang dikirimkan tersebut diterapkan pada database. Namun di sini kita bikin commit nya itu selalu true, agar kita gak perlu capek-capek run commit() pada setiap query.

Di sini kita buat database bernama myfirstdb biar gak perlu pakai database bawaan postgres

Lanjut kita lakukan koneksi ulang ke database tersebut dengan menutup koneksi saat ini, serta membuat koneksi yang baru.

Berikutnya di sini gw punya beberapa csv untuk demo yang mau gw load ke database, yaitu customers, orders, products, dan sales. Kek nya gak perlu gw show 1/1 csv nya wkwkw.

Dengan mengikuti csv nya, kita lakukan database modeling, sesuaikan kolom dan atribut nya untuk tabel si customers, begitu juga dengan yang lainnya.

Sekarang di tahapan load data, kalau coba kita bayangin, masa kita perlu nge query "INSERT INTO customers (kolom1, kolom2, kolom3) VALUES (nilai1, nilai2, nilai3) untuk semua baris? Jadi di sini kita pakai teknik for loop untuk dapetin semua nilai di setiap baris kemudian kita eksekusi query insert into nya.

Tahapan yang dilakukan adalah membaca file csv terlebih dahulu, kasih mode nya read dan parameter lainnya (kita namakan objek nya file). Berikutnya, file itu klo di print masih jelek, gak kek csv gitu, jadi perlu dibaca sebagai csv dengan cara kita buat variabel baru kemudian assign class csv.reader yang membaca file. Terakhir, tinggal dipanggil setiap row dari variabel yang udah bisa baca csv nya, dan tinggal execute setiap barisnya pakai for each.

Next kita bisa pakai perintah fetch untuk ngambil output yang dikeluarkan database ketika kita melakukan query menggunakan cursor. fetch ini ada 3 macam, fetchone, fetchmany, dan fetchall. Dah kebayang lah yak bedanya apa, intinya fetchmany bisa nentuin mau show berapa (parameter: int). Kalau kita lihat ss di atas, kenapa pakai fetchone tapi malah show semua? Karena kita pakai iterasi, untuk ngeprint semua output cursor nya sampai habis

Ketika kita melakukan fetchall, maka semua barisnya ditampilkan di console kita, lihat ada tanda list sebelum tuple [(0, 'Shirt' ...] yang berarti dia ngambil banyak tuple dan disimpan dalam 1 list (kita bahas ini nanti). Ketika kita lakukan fetchall(), dan coba fetchone() sekali lagi, maka hasilnya adalah None dan code while pada ss di atas akan terhenti karena baris keluarannya sudah habis.

Ini analisis kecil gw, dan ini gw tanyain ke ChatGPT ngapain kita pakai while trus pakai fetchone()? Kenapa gak langsung fetchall() aja?

Trus gw coba bandingin bikin variabel yang sama dengan nama fetch nya, dan ngeliat size nya

Disclaimer, gw rasa size ini cuma representatif kolom dan baris yang ditampilkan, tapi harusnya ya gitu, fetchall akan menggunakan memori lebih besar karena menampung banyak data, maka penggunaan while dan fetchone akan lebih diunggulkan untuk optimasi penggunaan memori.