Datatransformaties in Databricks

Databricks-concepten

Kevin Barlow

Data Practitioner

SQL voor data-engineering

SQL

Vertrouwd voor databasebeheerders (DBA's)
Geschikt voor standaardbewerkingen
Voer vooraf gedefinieerde UDF's uit

-- Creating a new table in SQL

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

Andere talen voor data-engineering

Python, R, Scala

Vertrouwd voor software-engineers
Standaard en complexe transformaties
Eigen functies gebruiken en definiëren

#Creating a new table in Pyspark

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

Veelvoorkomende transformaties

Schema-manipulatie

Kolommen toevoegen en verwijderen
Kolommen herdefiniëren

#Pyspark

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

Filteren

Beperk DataFrame tot subset
Meerdere criteria doorgeven

#Pyspark

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

Veelvoorkomende transformaties (vervolg)

Geneste data

Array- of Struct-data
Uitvouwen of samenvoegen

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

Aggregatie

Groepeer op kolommen
Bereken samenvattingen

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

Auto Loader

Auto Loader verwerkt nieuwe databestanden zodra ze in de data lake landen.

Incrementele verwerking
Efficiënte verwerking
Automatisch

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

Auto Loader-diagram

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

Structured Streaming

Streaming-pijplijn

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

Laten we oefenen!

Databricks-concepten