Databricks’te veri dönüşümleri

Databricks Kavramları

Kevin Barlow

Data Practitioner

Veri mühendisliği için SQL

SQL

Veritabanı Yöneticilerine (DBA) tanıdık
Standart işlemler için ideal
Ön tanımlı UDF’leri çalıştırır

-- Creating a new table in SQL

CREATE TABLE table_name
USING delta
AS (
  SELECT *
  FROM source_table
  WHERE date >= '2023-01-01'
)

Veri mühendisliği için diğer diller

Python, R, Scala

Yazılım mühendislerine tanıdık
Standart ve karmaşık dönüşümler
Özel fonksiyonları kullan ve tanımla

#Creating a new table in Pyspark

spark
  .read
  .table('source_table')
  .filter(col('date') >= '2023-01-01')
  .write
  .saveAsTable('table_name')

Yaygın dönüşümler

Şema düzenleme

Sütun ekle veya kaldır
Sütunları yeniden tanımla

#Pyspark

df
  .withColumn(col('newCol'), ...)
  .drop(col('oldCol'))

Filtreleme

DataFrame’i alt kümeye indir
Birden çok ölçüt geç

#Pyspark

df
  .filter(col('date') >= target_date)
  .filter(col('id') IS NOT NULL)

Yaygın dönüşümler (devam)

İçiçe veri

Dizi (Array) veya Struct verisi
Genişlet veya daralt

df
  .explode(col('arrayCol')) #wide to long
  .flatten(col('items')) #long to wide

Toplulaştırma

Sütunlara göre grupla
Özet değerleri hesapla

df
  .groupBy(col('region'))
  .agg(sum(col('sales')))

Auto Loader

Auto Loader, veri gölüne inen yeni dosyaları işler.

Artımlı işleme
Verimli işleme
Otomatik

spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .load(file_path)

Auto Loader Diyagramı

¹ https://www.databricks.com/blog/2020/02/24/introducing-databricks-ingest-easy-data-ingestion-into-delta-lake.html

Structured Streaming

Akış İşlem Hattı

spark.readStream
    .format("kafka")
    .option("subscribe", "<topic>")
    .load()
    .join(table_df, 
      on="<id>", how="left")
    .writeStream
    .format("kafka")
    .option("topic", "<topic>")
    .start()

Haydi pratik yapalım!

Databricks Kavramları