Transformasi data di Databricks

Konsep Databricks

Kevin Barlow

Data Practitioner

SQL untuk rekayasa data

SQL

Familiar bagi Database Administrator (DBA)
Cocok untuk manipulasi standar
Menjalankan UDF terdefinisi

-- Creating a new table in SQL

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

Bahasa lain untuk rekayasa data

Python, R, Scala

Familiar bagi software engineer
Transformasi standar dan kompleks
Gunakan dan definisikan fungsi kustom

#Creating a new table in Pyspark

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

Transformasi umum

Manipulasi skema

Tambah dan hapus kolom
Ubah definisi kolom

#Pyspark

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

Penyaringan

Kurangi DataFrame ke subset data
Gunakan banyak kriteria

#Pyspark

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

Transformasi umum (lanjutan)

Data bertingkat (nested)

Data Array atau Struct
Bentangkan atau ringkas

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

Agregasi

Kelompokkan data berdasarkan kolom
Hitung ringkasan data

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

Auto Loader

Auto Loader memproses file data baru saat masuk ke data lake.

Pemrosesan inkremental
Pemrosesan efisien
Otomatis

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

Diagram Auto Loader

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

Structured Streaming

Pipa Streaming

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

Ayo berlatih!

Konsep Databricks