การแปลงข้อมูลใน Databricks

แนวคิด Databricks

Kevin Barlow

Data Practitioner

SQL สำหรับ Data Engineering

SQL

คุ้นเคยสำหรับ Database Administrators (DBAs)
เหมาะกับการจัดการข้อมูลมาตรฐาน
รัน UDF ที่กำหนดไว้ล่วงหน้าได้

-- Creating a new table in SQL

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

ภาษาอื่น ๆ สำหรับ Data Engineering

Python, R, Scala

คุ้นเคยสำหรับนักพัฒนาซอฟต์แวร์
รองรับการแปลงข้อมูลทั้งแบบมาตรฐานและซับซ้อน
ใช้และกำหนดฟังก์ชันเองได้

#Creating a new table in Pyspark

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

การแปลงข้อมูลทั่วไป

การจัดการ Schema

เพิ่มและลบคอลัมน์
กำหนดคอลัมน์ใหม่

#Pyspark

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

การกรองข้อมูล

ลด DataFrame ให้เหลือเฉพาะข้อมูลที่ต้องการ
กำหนดเงื่อนไขหลายข้อพร้อมกันได้

#Pyspark

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

การแปลงข้อมูลทั่วไป (ต่อ)

ข้อมูลแบบซ้อน

ข้อมูลประเภท Array หรือ Struct
ขยายหรือบีบอัดโครงสร้างได้

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

การรวมกลุ่มข้อมูล

จัดกลุ่มข้อมูลตามคอลัมน์
คำนวณค่าสรุปข้อมูล

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

Auto Loader

Auto Loader ประมวลผลไฟล์ข้อมูลใหม่ทันทีที่ถูกนำเข้า data lake

ประมวลผลแบบเพิ่มทีละส่วน
ประมวลผลได้อย่างมีประสิทธิภาพ
ทำงานอัตโนมัติ

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

ไดอะแกรม Auto Loader

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

Structured Streaming

Streaming Pipeline

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

มาฝึกกันเถอะ!

แนวคิด Databricks