Transformations de données dans Databricks

Concepts de Databricks

Kevin Barlow

Data Practitioner

SQL pour l'ingénierie des données

SQL

Familier pour les administrateurs de bases de données (DBA)
Idéal pour les manipulations standard
Exécuter des UDF prédéfinies

-- Creating a new table in SQL

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

Autres langages pour l'ingénierie des données

Python, R, Scala

Familier pour les ingénieurs logiciels
Transformations standard et complexes
Utiliser et définir des fonctions personnalisées

#Creating a new table in Pyspark

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

Transformations courantes

Manipulation de schéma

Ajouter et supprimer des colonnes
Redéfinir des colonnes

#Pyspark

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

Filtrage

Réduire le DataFrame à un sous-ensemble de données
Appliquer plusieurs critères

#Pyspark

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

Transformations courantes (suite)

Données imbriquées

Données de type Array ou Struct
Déplier ou regrouper

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

Agrégation

Regrouper selon des colonnes
Calculer des résumés de données

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

Auto Loader

Auto Loader traite les nouveaux fichiers de données dès leur arrivée dans un lac de données.

Traitement incrémentiel
Traitement efficace
Automatique

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

Schéma d'Auto Loader

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

Structured Streaming

Pipeline de diffusion en continu

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

Passons à la pratique !

Concepts de Databricks