Snowflake’te Veri Boru Hatları Oluşturma: Yeni Başlayanlar İçin Rehber

Veri boru hatları, veri mühendisliğinin temel bileşenidir. Ham veriyi kaynak sistemlerden veri ambarına taşıyan adımlar dizisini temsil ederler; burada veriler temizlenir, dönüştürülür ve analize hazır hale getirilir. İyi tasarlanmış bir boru hattı, verinin alımdan içgörüye kadar verimli, güvenilir ve güvenli bir şekilde akmasını sağlar.

Bulut tabanlı güçlü bir veri platformu olan Snowflake, ölçeklenebilir, esnek ve maliyet etkin veri boru hatları kurmak için özellikle uygundur. Veri yükleme, dönüşüm, otomasyon ve izleme için yerleşik özellikleri sayesinde Snowflake, veri ekiplerinin karmaşık iş akışlarını sadeleştirip veriden değer üretmeye odaklanmasına olanak tanır.

Bu eğitimde, Snowflake’te bir veri boru hattı oluşturma sürecini—ortamınızı kurmaktan Streams ve Tasks gibi yerel araçları kullanarak veri dönüşümlerini otomatikleştirmeye kadar—adım adım ele alacağız. Bazı adımlar, bir bulut hizmetinin entegrasyon süreci ve olay yönetim sistemlerine dair bilgi gerektirebilir. Bunlar için kullandığınız bulut hizmetinin belgelerini öneririm.

Snowflake’e yeni başlıyorsanız, uygulamalı öğrenmek ve verileri sorgulama, modelleme ve yönetme becerilerini edinmek için Snowflake Foundations beceri yolumuzu değerlendirebilirsiniz. 

Snowflake’in Mimarisi Anlamak

Snowflake’in benzersiz mimarisi, veri boru hatlarını desteklemek üzere iyi tasarlanmıştır. Snowflake, işlem ve depolamayı birbirinden ayırır; bu da her birini bağımsız olarak ölçekleyebileceğiniz anlamına gelir. 

Veriler bulut depolama katmanında merkezi olarak saklanırken, hesaplama ihtiyaca göre başlatılıp durdurulabilen ve yeniden boyutlandırılabilen sanal ambarlarda gerçekleşir.

Bu mimari, birden fazla kullanıcı ve sürecin aynı veriye çakışma olmadan erişmesine olanak tanır; bu da alım, dönüşüm ve analiz içeren boru hattı iş yükleri için idealdir. 

Ayrıca, veri değişikliklerinin kullanıcıların o veriyi sorgulama yeteneğini doğrudan etkilemeden gerçekleşebilmesini sağlayan bir güvenlik katmanı ekler. 

Daha fazla bilgi ve burada ele alınmayan bulut katmanını öğrenmek için şu Snowflake mimarisine derinlemesine bakışı öneririm.

Snowflake boru hatları için temel bileşenler

Her boru hattında gerekli adımları göz önünde bulundurmamız gerekir. Genellikle, verileri Snowflake’e yüklemeden önce bir aşama alanında (stage) bekletirsiniz. Snowflake, tablolar ve şemaları dahil olmak üzere veri depolamayı yönetirken, sanal ambarlar sorguları ve diğer veri işlemlerini yürütmek için gereken hesaplama kaynaklarını sağlar.

İşte Snowflake veri boru hatlarında kullanılan bu temel bileşenlerin net tanımları:

• Stages: Verileri yüklemek/boşaltmak için kullanılan geçici veya harici depolama konumları.
• Virtual Warehouses: Özellikle Snowflake’in sağladığı hesaplama kaynaklarını ifade eder. Sanal ambarlar, birden fazla Sanal Makineden (VM) oluşan Yoğun Paralel İşleme (MPP) kümeleridir. Snowflake, sorguları işlemek ve veri dönüşümlerini verimli şekilde gerçekleştirmek için bu kaynakları sağlar ve yönetir.
• Databases: Verilerinizi düzenlemek için mantıksal kaplar. Genellikle bir tablo koleksiyonunu barındırırlar.
• Schemas: İlgili tabloları ve nesneleri gruplamak için veritabanı içindeki alt bölümler. Tabloların birbirleriyle nasıl ilişkili olduğunu tanımlar.
• Tables: Yapılandırılmış verilerin saklandığı yerler.

Snowflake Ortamınızı Kurma

Snowflake ortamınızın hazır olduğundan emin olalım. Snowflake’e yeniyseniz, başlamak için şu adımları izleyin:

Snowflake hesabı oluşturun

Bir Snowflake hesabınız yoksa, 30 günlük ücretsiz deneme süresi mevcut olup (bu yazının yazıldığı tarihte) 400 $’lık ücretsiz kullanım içerir. Bu, birkaç örnek boru hattını test etmek için yeterli olacaktır!

1. Snowflake Trial adresini ziyaret edin ve ücretsiz bir hesaba kaydolun.

Snowflake deneme sürümü kayıt formu

2. Amacımız doğrultusunda, Snowflake’i kullanmaya devam ederseniz gelecekte kredi maliyetlerini azaltmak için Enterprise sürümü yerine Standard sürümünü seçeceğiz!
3. Tercih ettiğiniz bulut sağlayıcısını seçin (AWS, Azure veya GCP). Bu eğitimdeki örnekler AWS kullanacaktır.
4. Otomatik seçilen veri merkezi, coğrafi olarak size en yakın olanı olmalı ve en iyi deneyimi sağlamalıdır.
5. Hesap oluşturduktan sonra Snowflake Web UI’ına giriş yapın.

Arayüzde gezinme

İçeri girdikten sonra şunları göreceksiniz:

• Worksheet: SQL sorgularını çalıştırabileceğiniz yer.
• Databases: Veritabanlarını ve şemaları görüntüleyin ve yönetin.
• Warehouses: Hesaplama kaynaklarını izleyin ve kontrol edin.
• History: Daha önce çalıştırılan sorguları ve görevleri takip edin

Snowflake kullanımı hakkında daha fazla tavsiye isterseniz bu Snowflake Foundations kursunu takip edin.

Gerekli kaynakları oluşturma

Bu adımda bir sanal ambar, veritabanı ve şema oluşturmayı ele alacağız. Ortamınızı kurmak için çalışma sayfanızda aşağıdaki SQL’i çalıştırın:

-- 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;Altyapı:

Verileri Snowflake’e Yükleme

Snowflake’te Veri Alımı hakkındaki makalemizde tartıştığımız gibi, Snowflake hem toplu (bulk) hem de sürekli yüklemeyi destekler. Toplu yükleme, bir seferde tablonuza büyük veri parçaları yüklemek için tasarlanmıştır. Bu genellikle planlı aralıklarla yapılır veya Snowpipe’ın olaya dayalı yönetimiyle otomatikleştirilir. 

Sürekli (akış) yükleme, Snowpipe Streaming gibi bileşenlerle gerçekleştirilebilir. Veriyi içeri almak için bulut sağlayıcımızla veri entegrasyonları (stage’ler) oluşturmalı ve COPY INTO komutunu kullanarak veriyi yüklemeliyiz.

S3 stage ve entegrasyonu oluşturma

Veri alımı, herhangi bir boru hattının ilk adımıdır. Bu adımda bir depolama entegrasyonu ve bir stage oluşturmanın temelini ele alacağız. Bunu yapmak için, Snowflake’inizi bir Snowflake IAM kullanıcısı ile AWS’ye zaten bağladığınızı varsayalım. Aksi halde, dokümantasyondaki Snowflake AWS yapılandırma kılavuzuna bakın.

-- 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;Altyapı:

Sonunda, AWS ile Snowflake entegrasyonunuzu ve stage’inizi başarıyla oluşturmuş olmalısınız. AWS temellerini gözden geçirmek isterseniz, aşağıdaki makale harika bir AWS’ye giriş kaynağıdır.

COPY INTO ile toplu yükleme

COPY INTO komutu, verileri Snowflake’e yüklemenin yaygın yöntemidir. Verileri Snowflake’e yerel olarak yüklenmiş bir dosyadan ya da bulut bağlantılarımızı kullanarak alabiliriz. Aşağıdaki örnekte, bir CSV dosyasını kopyalamak için stage’imizi kullanacağız.

-- 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');Altyapı:

Snowpipe ile yüklemeyi otomatikleştirme

Her yeni dosya geldiğinde elle yükleme yapmak, hız arttıkça verimsiz ve zaman alıcıdır. Bunun yerine Snowpipe kullanarak stage’imizdeki dosyaları yükleyebiliriz. Ardından AWS olay bildirimlerini Snowflake IAM kullanıcımıza bağlayarak her seferinde Snowpipe’ın tetiklenmesini sağlarız.

-- 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');Altyapı:

Sonraki adımda, hedef kovamıza bir dosya geldiğinde IAM kullanıcımıza bildirim göndermek için Amazon SQS gibi bir hizmet kullanabiliriz.

Son olarak Snowpipe’ı şu sorguyla izleyebiliriz:

SELECT * FROM INFORMATION_SCHEMA.PIPE_USAGE_HISTORY;Altyapı:

Artık S3 kovunuza her dosya geldiğinde verileri yükleme konusunda endişelenmenize gerek yok! Bu bulut bildirimleri, Snowpipe’ınızı verileri yüklemek üzere tetikleyecek. 

Streams ve Tasks ile Verileri Dönüştürme

Snowflake’in Streams ve Tasks özellikleri, değişen verileri izlemek ve dönüştürmek için yerleşik araçlardır. Streams, tablolardaki değişiklikleri izlemeye odaklanırken Tasks, SQL ifadelerinin yürütülmesini otomatikleştirmek için kullanılır.

Streams’e giriş

Bir Stream, bir tablodaki Değişiklik Verisi Yakalama (CDC) bilgisini—yani eklemeler, güncellemeler ve silmeler—son sorgulamadan bu yana yakalar. Bunlar genellikle Veri İşleme Dili (DML) sorguları biçimindedir. Bu sayede, iki farklı sorgulama zamanı arasındaki tablo değişikliklerini görebiliriz.

-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;Altyapı:

Daha sonra yalnızca değişiklikleri almak için stream’i sorgulayabilirsiniz:

SELECT * FROM my_stream WHERE METADATA$ACTION = 'INSERT';Altyapı:

Tasks’e giriş

Tasks, SQL yürütmesini bir zamanlamaya veya olay tetikleyicisine bağlı olarak otomatikleştirmenizi sağlar. Görevler; tablolarımız ve boru hatlarımız üzerinde prosedürleri otomatikleştirmek için SQL, Javascript, Python vb. dillerdeki kodları çağırabilir.

-- 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;Altyapı:

Streams ve Tasks’i birleştirme

Bu, otomatik dönüşüm boru hatlarının temelini oluşturur. Yukarıda gördüğünüz gibi, Tasks yeni verileri farklı bir tabloya çekmek için Streams’i kullanabilir. Genel süreç şu şekilde olabilir:

• Bir veri tablosundaki değişiklikleri izlemek için bir Stream oluşturun
• Belirli aralıklarla bu stream’den veri çeken bir Task oluşturun
• Her veri çekişinde veriyi istenen formata dönüştürün
• Bu veriyi hedef tabloya yükleyin

Bu şekilde, bir veri boru hattının parçalarını oluşturmuş oluruz. Streams ve Tasks’i birleştirerek yalnızca yeni yüklenen veriler üzerinde dönüşüm gerçekleştirebiliriz. Yalnızca Tasks kullansaydık, her yeni veri istediğimizde ham tablolarımızın tamamını işlemi olmamız gerekebilirdi; bu da maliyetli ve hesaplama gücünün verimsiz kullanımı olurdu.

Uçtan Uca Bir Veri Boru Hattı Oluşturma

Basit bir boru hattı kullanım senaryosunu inceleyelim: sipariş verilerini içeri almak, değişiklikleri izlemek ve analitik için dönüştürmek. Her adımı ayrı ayrı nasıl yürüteceğimizi ele aldığımıza göre, bu bölümde adımların bir arada nasıl görünebileceğine değineceğiz. 

Adım 1: Bulut depolama entegrasyonu ve stage oluşturun

Öncelikle Snowflake ekosistemimizi bulut depolamamıza bağlamalıyız. Örnek olarak AWS kullandık. Snowflake’te CREATE STORAGE INTEGRATION ve CREATE STAGE komutlarını kullanarak AWS ile veritabanımız arasında bağlantı kurduk.

-- 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;Altyapı:

Adım 2: Snowpipe ile verileri içeri alın

İkinci adımda, ham veri tablolarımıza COPY INTO komutunu çalıştıracak bir Snowpipe oluşturuyoruz. 

-- 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');Altyapı:

Adım 3: Değişiklikleri yakalamak için Streams kullanın

Daha sonra, bu ham veri tablosu üzerinde değişiklikleri izlemek için bir stream kullanıyoruz.

-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;Altyapı:

Adım 4: Tasks ile dönüşümleri otomatikleştirin

Sonraki adımda, bu yeni yüklenen verileri işlenmiş tablo sürümümüze dönüştürüp yükleyen bir Task oluşturuyoruz. Bu görev, yeni veri olup olmadığını kontrol etmek için bir zamanlamayla tetiklenecek. Yeni veri yoksa Task çalışmayacaktır. 

-- 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;Altyapı:

Adım 5: İzleyin ve yönetin

Son olarak, boru hatlarımızın amaçlandığı gibi çalıştığından emin olmak isteriz. SHOW PIPES ve SHOW TASKS komutlarını kullanarak bunların doğru şekilde kurulduğunu ve çalıştığını doğrulayın. Ardından, veri yükleme performansını ve hataları takip etmek için INFORMATION_SCHEMA’daki meta veri tablolarını kullanın.

Snowflake’te Veri Boru Hattı Geliştirme İçin En İyi Uygulamalar

Her zaman olduğu gibi, geliştirmede en iyi uygulamalara uygun şekilde çalıştığımızdan emin olmak isteriz. Dikkate alınacak konular arasında performans ve maliyet için optimizasyon, veri kalitesi ve güvenilirliğini sağlama ile uygun güvenlik ve erişim düzeylerini uygulama bulunur.

Performans ve maliyet için optimize edin

Amacımız, boru hatlarımızı her zaman mümkün olan en verimli ve en düşük maliyetle işletmek olmalıdır. Böylelikle sunucular üzerindeki etkimizi en aza indirir ve veriyi olabildiğince zamanında sunarız. İşte bazı ipuçları: 

• Bekleme ücretlerinden kaçınmak için ambarlarda otomatik askıya almayı kullanın. Ambarlar zaman bazlı ücretlendirdiği için, bu yöntem hesaplama maliyetlerini düşürür.
• Sıkıştırılmış dosyalar yükleyin ve bunları en uygun boyutlarda gruplayın. Çok küçük dosyalar, her dosyayı yüklemek için bağlantı oluşturma gerektirdiğinden fazla ağ genel gideri yaratır. Çok büyük dosyalar ise fazla bant genişliği tüketebilir ve başarısızlık riski taşır. Snowflake, 100–250 MB boyutunda deltalar önerir.
• Dönüşümleri düşük kullanım pencerelerinde zamanlayın. Böylece kaynaklar için rekabet etmezsiniz.

Veri kalitesini ve güvenilirliğini sağlayın

Veri kalitesi ve güvenilirliği, her veri mühendisi için kritiktir. Yüklenen verinin hatasız ve kullanılabilir olduğundan emin olmak istersiniz. Ayrıca sorunlar yaşandığında veri boru hattınızın farkında olmadan veri kaybına yol açmamasını istersiniz. Veri kalitesi yönetimi için bazı ipuçları:

• Alımdan önce ve sonra veriyi doğrulayın. Bu, ara tabloların periyodik kontrolleri ve uyarılarla yapılabilir.
• Sorunlu kayıtları izole etmek ve düzeltmek için stream’leri ve ara tabloları kullanın. Yalnızca değişen veriler çekildiği için, stream’lerle sorunlarımızın kapsamını en aza indirebiliriz.
• Gerektiğinde benzersiz kısıtlar uygulayın veya çift kayıtları temizleme mantığı ekleyin. Bu, örneğin NULL verilerin kabul edilemez olduğu alanlarda bunu engelleyebilir ya da örtüşen veri çekimlerinde önceden çift kayıtları ayıklayabilir.

Güvenlik ve erişim kontrollerini uygulayın

Son olarak, verinize uygun erişim ve güvenliği sağlamanız önemlidir. Bazı tablolar hassas bilgiler barındırabilir ya da yalnızca belirli bir grubun değiştirmesine izin verilmelidir. Snowflake’te güvenlik ve erişimi doğru şekilde yönetmek için şu önerileri izleyin:

• Erişimi kısıtlamak için roller ve yetkiler kullanın.
• Hassas verileri sütun düzeyinde maskeleme veya satır erişim ilkeleri ile koruyun.
• Depolama kimlik bilgilerini periyodik olarak yenileyin ve stage kullanımını izleyin.

Sonuç

Snowflake’te veri boru hatları kurmak, veri mühendisliği dünyasına adım atan yeni başlayanlar için yalın ve ölçeklenebilir bir süreçtir. Sezgisel mimarisi, güçlü alım seçenekleri ve Streams ile Tasks gibi yerleşik otomasyon araçlarıyla Snowflake, veri ekiplerinin ham veriden eyleme dönük içgörülere hızla ilerlemesini sağlar.

Bu rehberi izleyerek artık Snowflake’te kendi boru hattınızı kurmak için temel bilgiye sahipsiniz—alım, dönüşüm ve orkestrasyonu tek bir platformda ele alabilirsiniz. Snowflake hakkında daha fazla bilgi için şu rehberlere göz atın:

• Yeni Başlayanlar için Snowflake
• Snowflake’e Giriş Kursu
• Snowflake Nasıl Öğrenilir
• Snowflake’te Veri Modellemesi
• En İyi Snowflake Sertifikası 2025