Transformace dat v Databricks

Databricks Concepts

Kevin Barlow

Data Practitioner

SQL pro datové inženýrství

SQL

Známé pro správce databází (DBA)
Vhodné pro standardní manipulace
Spouštění předdefinovaných UDF

-- Creating a new table in SQL

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

Další jazyky pro datové inženýrství

Python, R, Scala

Známé pro softwarové inženýry
Standardní i komplexní transformace
Použití a definice vlastních funkcí

#Creating a new table in Pyspark

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

Běžné transformace

Manipulace se schématem

Přidání a odebrání sloupců
Předefinování sloupců

#Pyspark

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

Filtrování

Omezení DataFrame na podmnožinu dat
Předání více kritérií

#Pyspark

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

Běžné transformace (pokračování)

Vnořená data

Pole nebo strukturovaná data
Rozbalení nebo sbalení

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

Agregace

Seskupení dat podle sloupců
Výpočet souhrnů dat

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

Auto Loader

Auto Loader zpracovává nové soubory při jejich příchodu do datového jezera.

Inkrementální zpracování
Efektivní zpracování
Automatické

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

Diagram Auto Loaderu

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

Structured Streaming

Diagram streamovacího pipeline

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

Pojďme cvičit!

Databricks Concepts