Membangun Data Pipeline di Snowflake: Panduan Ramah Pemula

Data pipeline adalah komponen inti dari rekayasa data. Ini mewakili rangkaian langkah yang memindahkan data mentah dari sistem sumber ke gudang data, tempat data dapat dibersihkan, ditransformasikan, dan disiapkan untuk dianalisis. Pipeline yang dirancang dengan baik memastikan aliran data berjalan efisien, andal, dan aman dari proses ingestion hingga menghasilkan insight.

Snowflake, sebuah platform data berbasis cloud yang kuat, sangat cocok untuk membangun data pipeline yang skalabel, fleksibel, dan hemat biaya. Dengan fitur bawaan untuk pemuatan data, transformasi, otomatisasi, dan pemantauan, Snowflake memungkinkan tim data menyederhanakan alur kerja yang kompleks dan fokus pada penyampaian nilai dari data.

Dalam tutorial ini, kami akan membahas proses membangun data pipeline di Snowflake—mulai dari menyiapkan lingkungan Anda hingga mengotomatisasi transformasi data menggunakan alat native seperti Streams dan Tasks. Beberapa bagian akan membutuhkan pengetahuan tentang proses integrasi layanan cloud dan sistem manajemen event. Untuk hal tersebut, saya merekomendasikan dokumentasi dari layanan cloud yang Anda gunakan.

Jika Anda baru memulai dengan Snowflake, pertimbangkan alur keahlian Snowflake Foundations kami untuk praktik langsung dan mempelajari cara melakukan query, memodelkan, serta mengelola data. 

Memahami Arsitektur Snowflake

Arsitektur unik Snowflake dirancang dengan baik untuk mendukung data pipeline. Snowflake memisahkan komputasi dan penyimpanan, artinya Anda dapat menskalakan masing-masing secara independen. 

Data disimpan secara terpusat di lapisan penyimpanan cloud, sementara komputasi terjadi di virtual warehouse, yang dapat dijalankan, dihentikan, dan diubah ukurannya sesuai kebutuhan.

Arsitektur ini memungkinkan banyak pengguna dan proses mengakses data yang sama tanpa saling berebut sumber daya, sehingga ideal untuk beban kerja pipeline yang melibatkan ingestion, transformasi, dan analisis. 

Ini juga menambah lapisan keamanan di mana perubahan data dapat terjadi tanpa langsung memengaruhi kemampuan pengguna untuk melakukan query terhadap data tersebut. 

Untuk informasi lebih lanjut dan mempelajari tentang lapisan cloud yang tidak dibahas di sini, saya merekomendasikan pembahasan mendalam tentang arsitektur Snowflake ini.

Komponen kunci untuk pipeline Snowflake

Di setiap pipeline, kita harus mempertimbangkan langkah-langkah yang diperlukan. Biasanya, Anda akan melakukan staging data sebelum memuatnya ke Snowflake. Snowflake sendiri menangani penyimpanan data, termasuk tabel dan skemanya, sementara virtual warehouse menyediakan sumber daya komputasi yang dibutuhkan untuk mengeksekusi query dan operasi data lainnya.

Berikut adalah definisi jelas dari komponen esensial yang digunakan dalam data pipeline Snowflake:

• Stage: Lokasi penyimpanan sementara atau eksternal yang digunakan untuk memuat/menurunkan data.
• Virtual Warehouse: Mengacu secara spesifik pada sumber daya komputasi yang disediakan oleh Snowflake. Virtual warehouse pada dasarnya adalah klaster komputasi Massively Parallel Processing (MPP), yang terdiri dari beberapa Virtual Machine (VM). Snowflake menyediakan dan mengelola sumber daya ini untuk memproses query dan melakukan transformasi data secara efisien.
• Database: Wadah logis untuk mengatur data Anda. Ini sering kali menyimpan sekumpulan tabel.
• Skema: Subdivisi dalam sebuah database untuk mengelompokkan tabel dan objek terkait. Ini mendefinisikan bagaimana tabel-tabel saling berhubungan.
• Tabel: Tempat penyimpanan data terstruktur.

Menyiapkan Lingkungan Snowflake Anda

Mari pastikan Anda memiliki lingkungan Snowflake yang siap digunakan. Jika Anda baru dengan Snowflake, ikuti langkah-langkah berikut untuk memulai:

Buat akun Snowflake

Jika Anda belum memiliki akun Snowflake, tersedia uji coba gratis 30 hari yang mencakup (pada saat penulisan) penggunaan gratis senilai $400. Ini seharusnya cukup untuk menguji beberapa pipeline contoh!

1. Kunjungi Snowflake Trial dan daftar untuk akun gratis.

Formulir pendaftaran untuk Snowflake Trial

2. Untuk keperluan kita, pilih edisi Standard, bukan edisi Enterprise, untuk menghemat biaya kredit di masa depan jika Anda melanjutkan menggunakan Snowflake!
3. Pilih penyedia cloud yang Anda sukai (AWS, Azure, atau GCP). Contoh dalam tutorial ini akan menggunakan AWS.
4. Pusat data yang dipilih secara otomatis seharusnya yang paling dekat dengan lokasi Anda dan memberikan pengalaman terbaik.
5. Setelah membuat akun, masuk ke Snowflake Web UI.

Menavigasi UI

Setelah masuk, Anda akan melihat:

• Worksheet: Tempat Anda dapat menjalankan query SQL.
• Database: Melihat dan mengelola database dan skema.
• Warehouse: Memantau dan mengendalikan sumber daya komputasi.
• History: Melacak query dan tugas yang pernah dijalankan

Jika Anda menginginkan saran lebih lanjut tentang penggunaan Snowflake, ikuti kursus Snowflake Foundations ini.

Membuat sumber daya yang diperlukan

Langkah ini mencakup pembuatan virtual warehouse, database, dan skema. Jalankan SQL berikut di worksheet Anda untuk menyiapkan lingkungan:

-- set the user role appropriately
USE ROLE accountadmin;

-- Create a virtual warehouse
CREATE WAREHOUSE datacamp_pipeline_wh WITH WAREHOUSE_SIZE = 'XSMALL' AUTO_SUSPEND = 60;

-- Set our active working warehouse
USE WAREHOUSE datacamp_pipeline_wh;

-- create our Datacamp Sample Database
CREATE OR REPLACE DATABASE datacamp_sample_db;

-- create the Datacamp (Sample) Raw Schema
CREATE OR REPLACE SCHEMA datacamp_sample_db.raw_schema;Didukung oleh

Memuat Data ke Snowflake

Seperti yang kami bahas dalam artikel tentang Ingestion Data di Snowflake, Snowflake mendukung pemuatan batch (massal) dan berkelanjutan. Pemuatan massal dirancang untuk mengunggah potongan data besar ke tabel Anda sekaligus. Ini umumnya dilakukan pada interval terjadwal, atau diotomatisasi dengan manajemen berbasis event Snowpipe. 

Pemuatan berkelanjutan (stream) dapat dilakukan menggunakan, misalnya, Snowpipe Streaming. Untuk melakukan ingestion data, kita harus membuat integrasi data (stage) dengan penyedia cloud kita dan memuat data menggunakan COPY INTO.

Membuat stage dan integrasi S3

Ingestion data adalah langkah pertama dalam setiap pipeline. Pada langkah ini, kita akan membahas dasar-dasar membuat storage integration dan membuat stage. Untuk melakukannya, bayangkan kita sudah menghubungkan Snowflake Anda ke AWS dengan pengguna IAM Snowflake. Jika belum, lihat panduan konfigurasi Snowflake AWS di dokumentasi.

-- set Role Context
USE ROLE accountadmin;

-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;

-- this creates the storage integration 
-- The STORAGE_AWS_ROLE_ARN will come from the AWS configuration guide steps
-- We will allow access to all locations and block any with sensitive data
CREATE STORAGE INTEGRATION datacamp_aws
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<iam_role>'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);

-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format 
TYPE = 'CSV' 
FIELD_OPTIONALLY_ENCLOSED_BY = '"';

USE SCHEMA datacamp_sample_db.raw_schema;

--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
  STORAGE_INTEGRATION = datacamp_aws
  URL = 's3://bucket1/path1/'
  FILE_FORMAT = my_csv_format;Didukung oleh

Pada akhirnya, Anda seharusnya berhasil membuat integrasi Snowflake Anda dengan AWS dan membuat stage. Jika Anda ingin meninjau dasar-dasar AWS, artikel berikut adalah pengantar AWS yang bagus.

Pemuatan massal dengan COPY INTO

Perintah COPY INTO adalah metode umum untuk mengunggah data ke Snowflake. Kita dapat menggunakan file yang diunggah secara lokal di Snowflake, atau menggunakan koneksi cloud kita. Pada contoh berikut, kita akan menggunakan stage untuk menyalin file CSV.

-- Load data using COPY INTO
-- The following will attempt to load all file in the bucket
-- To specify the particular files you can use the FILES or PATTERN settings
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @my_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');Didukung oleh

Mengotomatisasi Pemuatan dengan Snowpipe

Memuat secara manual setiap kali kita menerima file baru menjadi tidak efisien dan memakan waktu seiring meningkatnya kecepatan. Sebagai gantinya, kita dapat menggunakan Snowpipe untuk memuat file dari stage kita. Lalu, kita menggabungkan notifikasi event AWS ke pengguna IAM Snowflake kita untuk memicu Snowpipe setiap saat.

-- Let’ s first create our Snowpipe
CREATE PIPE my_pipe AS
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @datacamp_s3_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');Didukung oleh

Selanjutnya, kita dapat menggunakan sesuatu seperti Amazon SQS untuk mengirim notifikasi ke pengguna IAM kita setiap kali bucket target menerima file.

Terakhir, kita dapat memantau Snowpipe kita menggunakan:

SELECT * FROM INFORMATION_SCHEMA.PIPE_USAGE_HISTORY;Didukung oleh

Kini Anda tidak perlu khawatir mengunggah data setiap kali file tiba di bucket S3 Anda! Notifikasi cloud ini akan memicu Snowpipe Anda untuk memuat data. 

Mentransformasi Data dengan Streams dan Tasks

Streams dan Tasks Snowflake adalah alat bawaan untuk melacak dan mentransformasi data yang berubah. Stream berfokus pada pelacakan perubahan pada tabel, sedangkan Task digunakan untuk mengotomatisasi eksekusi pernyataan SQL.

Pengantar Streams

Stream menangkap informasi Change Data Capture (CDC)—yaitu insert, update, dan delete—pada sebuah tabel sejak terakhir kali di-query. Ini sering berupa query Data Manipulation Language (DML). Hal ini memungkinkan kita melihat perubahan pada tabel antara dua waktu query yang berbeda.

-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;Didukung oleh

Anda kemudian dapat melakukan query pada stream untuk mendapatkan hanya perubahan:

SELECT * FROM my_stream WHERE METADATA$ACTION = 'INSERT';Didukung oleh

Pengantar Tasks

Task memungkinkan Anda mengotomatisasi eksekusi SQL berdasarkan jadwal atau pemicu event. Task dapat memanggil SQL, Javascript, Python, dan sebagainya, untuk mengotomatisasi prosedur pada tabel dan pipeline kita.

-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
  WAREHOUSE = datacamp_sample_db 
SCHEDULE = '5 MINUTE'
AS
  INSERT INTO raw_schema.aws_processed_data
  SELECT * FROM my_stream WHERE location = ‘USA’;

-- This will allow us to test the task
EXECUTE TASK process_changes_task;Didukung oleh

Menggabungkan Streams dan Tasks

Ini adalah fondasi untuk pipeline transformasi terotomatisasi. Seperti yang Anda lihat di atas, Task dapat menggunakan Stream untuk menarik data baru ke tabel lain. Proses umum mungkin terlihat seperti berikut:

• Buat Stream untuk memantau perubahan pada sebuah tabel data
• Buat Task yang menarik data dari stream tersebut secara terjadwal
• Pada setiap penarikan data, transformasikan data ke format yang diinginkan
• Muat data tersebut ke tabel target kita

Dengan cara ini, kita telah membuat bagian-bagian dari data pipeline. Dengan menggabungkan Streams dan Tasks, kita dapat melakukan transformasi hanya pada data yang baru dimuat. Jika hanya menggunakan Tasks, kita berpotensi harus memproses seluruh tabel mentah setiap kali menginginkan data baru—ini bisa mahal dan tidak efisien dalam penggunaan daya komputasi.

Membangun Data Pipeline End-to-End

Mari kita bahas contoh kasus pipeline sederhana: meng-ingest data pesanan, melacak perubahan, dan mentransformasikannya untuk analitik. Karena kita telah membahas cara menjalankan setiap langkah secara terpisah, bagian ini membahas seperti apa langkah-langkah tersebut jika digabungkan. 

Langkah 1: Buat integrasi penyimpanan cloud dan stage

Pertama, kita harus menghubungkan ekosistem Snowflake kita ke penyimpanan cloud. Kita menggunakan AWS sebagai contoh. Di Snowflake, kita menggunakan perintah CREATE STORAGE INTEGRATION dan CREATE STAGE untuk membangun koneksi antara AWS dan database kita.

-- set Role Context
USE ROLE accountadmin;

-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;

-- this creates the storage integration 
CREATE STORAGE INTEGRATION datacamp_aws
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<iam_role>'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);

-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format 
TYPE = 'CSV' 
FIELD_OPTIONALLY_ENCLOSED_BY = '"';

USE SCHEMA datacamp_sample_db.raw_schema;

--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
  STORAGE_INTEGRATION = datacamp_aws
  URL = 's3://bucket1/path1/'
  FILE_FORMAT = my_csv_format;Didukung oleh

Langkah 2: Ingestion data dengan Snowpipe

Pada langkah kedua, kita membuat Snowpipe yang akan menjalankan perintah COPY INTO ke dalam tabel data mentah kita. 

-- Let’ s first create our SnowpipeCREATE PIPE my_pipe AS

COPY INTO datacamp_sample_db.raw_schema.aws_first_upload

FROM @datacamp_s3_stage

FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');Didukung oleh

Langkah 3: Gunakan Streams untuk menangkap perubahan

Kemudian kita menggunakan stream pada tabel data mentah ini untuk melacak perubahan.

-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;Didukung oleh

Langkah 4: Otomatiskan transformasi dengan Tasks

Berikutnya, kita membuat Task yang mentransformasikan dan memuat data yang baru dimuat ini ke versi tabel yang sudah diproses. Ini akan dipicu sesuai jadwal untuk memeriksa data baru. Jika tidak ada data baru, Task tidak akan berjalan. 

-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
  WAREHOUSE = datacamp_sample_db 
SCHEDULE = '5 MINUTE'
AS
  INSERT INTO raw_schema.aws_processed_data
  SELECT * FROM my_stream WHERE location = ‘USA’;

-- This will allow us to test the task
EXECUTE TASK process_changes_task;Didukung oleh

Langkah 5: Pantau dan kelola

Terakhir, kita ingin memastikan pipeline berjalan sesuai yang diharapkan. Pastikan gunakan perintah SHOW PIPES dan SHOW TASKS untuk memverifikasi bahwa semuanya telah disetel dan berjalan dengan benar. Lalu gunakan tabel metadata dari INFORMATION_SCHEMA untuk melacak kinerja pemuatan data dan error.

Praktik Terbaik untuk Pengembangan Data Pipeline di Snowflake

Seperti biasa, kita ingin memastikan bahwa kita beroperasi dengan praktik terbaik dalam pengembangan. Hal-hal yang perlu dipertimbangkan adalah mengoptimalkan kinerja dan biaya, memastikan kualitas dan keandalan data, serta menerapkan tingkat keamanan dan akses yang sesuai.

Optimalkan kinerja dan biaya

Tujuan kita harus selalu mengoperasikan pipeline seefisien mungkin dengan biaya serendah mungkin. Dengan begitu, kita meminimalkan dampak pada server dan menyajikan data secepat mungkin. Berikut beberapa tips: 

• Gunakan auto-suspend pada warehouse untuk menghindari biaya idle. Ini mengurangi biaya komputasi dengan menghentikan warehouse saat tidak digunakan, karena penagihan berdasarkan waktu.
• Muat file terkompresi dan kelompokkan ke ukuran yang optimal. File yang terlalu kecil menimbulkan overhead jaringan karena perlu membuat koneksi untuk memuat tiap file. File yang terlalu besar dapat memakan bandwidth terlalu banyak dan berisiko gagal. Snowflake merekomendasikan delta berukuran 100–250 MB.
• Jadwalkan transformasi pada waktu penggunaan rendah. Ini berarti Anda tidak bersaing memperebutkan sumber daya.

Pastikan kualitas dan keandalan data

Kualitas dan keandalan data adalah hal krusial untuk dipertimbangkan setiap data engineer. Anda ingin memastikan data yang dimuat dapat digunakan tanpa error. Anda juga ingin memastikan data pipeline Anda tidak secara tidak sengaja menyebabkan data hilang saat terjadi masalah. Berikut beberapa tips manajemen kualitas data:

• Validasi data sebelum dan sesudah ingestion. Ini dapat dilakukan dengan pemeriksaan berkala pada tabel staging dan alert.
• Gunakan stream dan tabel staging untuk mengisolasi dan memperbaiki record bermasalah. Karena hanya data yang berubah yang ditarik, kita dapat meminimalkan cakupan masalah menggunakan stream.
• Terapkan constraint unik atau logika deduplikasi bila diperlukan. Ini dapat mencegah hal seperti data NULL di tempat yang tidak boleh ada, atau melakukan pra-deduplikasi jika terjadi tumpang tindih penarikan data.

Terapkan kontrol keamanan dan akses

Terakhir, penting untuk memastikan akses dan keamanan yang tepat untuk data Anda. Beberapa tabel dapat menyimpan informasi sensitif atau tidak boleh diubah kecuali oleh kelompok tertentu. Ikuti saran berikut untuk mengelola keamanan dan akses Anda di Snowflake:

• Gunakan role dan grant untuk membatasi akses.
• Lindungi data sensitif dengan masking tingkat kolom atau kebijakan akses baris.
• Putar kredensial penyimpanan dan pantau penggunaan stage.

Kesimpulan

Membangun data pipeline di Snowflake adalah proses yang ringkas dan skalabel, sempurna untuk pemula yang memasuki dunia rekayasa data. Dengan arsitektur yang intuitif, opsi ingestion yang tangguh, dan alat otomatisasi bawaan seperti Streams dan Tasks, Snowflake memberdayakan tim data untuk bergerak cepat dari data mentah ke insight yang dapat ditindaklanjuti.

Dengan mengikuti panduan ini, Anda kini memiliki pengetahuan dasar untuk menyiapkan pipeline Snowflake Anda sendiri—menangani ingestion, transformasi, dan orkestrasi, semuanya dalam satu platform. Untuk informasi lebih lanjut tentang Snowflake, lihat panduan berikut:

• Snowflake untuk Pemula
• Kursus Pengantar Snowflake
• Cara Belajar Snowflake
• Pemodelan Data di Snowflake
• Sertifikasi Snowflake Terbaik 2025