Chuyển đổi dữ liệu trong Databricks

Các khái niệm về Databricks

Kevin Barlow

Data Practitioner

SQL cho kỹ thuật dữ liệu

SQL

Quen thuộc với DBA
Tốt cho thao tác chuẩn
Chạy UDF dựng sẵn

-- Creating a new table in SQL

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

Ngôn ngữ khác cho kỹ thuật dữ liệu

Python, R, Scala

Quen thuộc với kỹ sư phần mềm
Biến đổi chuẩn và phức tạp
Dùng và định nghĩa hàm tùy chỉnh

#Creating a new table in Pyspark

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

Các phép biến đổi phổ biến

Thao tác schema

Thêm và xóa cột
Định nghĩa lại cột

#Pyspark

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

Lọc

Giảm DataFrame xuống tập con
Truyền nhiều tiêu chí

#Pyspark

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

Các phép biến đổi phổ biến (tiếp)

Dữ liệu lồng nhau

Dạng mảng hoặc Struct
Mở rộng hoặc thu gọn

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

Gộp (Aggregation)

Gom nhóm theo cột
Tính tóm tắt dữ liệu

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

Auto Loader

Auto Loader xử lý các tệp dữ liệu mới khi chúng đến data lake.

Xử lý tăng dần
Hiệu quả
Tự động

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

Sơ đồ Auto Loader

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

Structured Streaming

Pipeline 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!

Các khái niệm về Databricks