Tutorial: Crie um pipeline ETL com LakeFlow Pipelines

Este tutorial explica como criar e implementar um pipeline ETL (extração, transformação e carregamento) para orquestração de dados usando Lakeflow Pipelines e Auto Loader. Um pipeline ETL implementa as etapas para ler dados de sistemas de origem, transformar esses dados com base nos requisitos, como verificações de qualidade de dados e deduplicação de registros, e gravar os dados em um sistema de destino, como um data warehouse ou um data lake.

Neste tutorial, você usará o pipeline e Auto Loader para:

• Ingira dados de origem brutos em uma tabela de destino.
• Transformar os dados brutos de origem e gravar os dados transformados em duas visualizações materializadas de destino.
• Consulte os dados transformados.
• Automatize o ETL pipeline com um trabalho Databricks.

Para obter mais informações sobre pipelines e Auto Loader, consulte Spark Declarative Pipelines no Lakeflow e O que é o Auto Loader?

Requisitos

Para completar este tutorial, você deve atender aos seguintes requisitos:

• O senhor pode entrar em um site Databricks workspace.
• O senhor tem Unity Catalog habilitado para o seu workspace.
• Sua conta deve ter o compute serverless habilitado. LakeFlow Pipelines Serverless não estão disponíveis em todas as regiões do Workspace. Consulte Recursos com disponibilidade regional limitada para regiões disponíveis.
• Ter permissão para criar um recurso compute ou acesso a um recurso compute.
• Possuir permissões para criar um novo esquema em um catálogo. As permissões necessárias são USE CATALOG e CREATE SCHEMA.
• Possuir permissões para criar um novo volume em um esquema existente. As permissões necessárias são USE SCHEMA e CREATE VOLUME.

Sobre o dataset

O site dataset usado neste exemplo é um subconjunto do conjunto de dados Million Song, uma coleção de recursos e metadados para faixas de música contemporânea. Esse dataset está disponível no conjunto de dados de amostra incluído em seu Databricks workspace.

Etapa 1: Criar um pipeline

Primeiro, crie um pipeline ao definir os datasets em arquivos (chamados de código-fonte ) usando a sintaxe do pipeline. Cada arquivo de código-fonte pode conter apenas uma linguagem, mas você pode adicionar vários arquivos específicos de linguagem ao pipeline. Para saber mais, consulte Spark Declarative Pipelines no Lakeflow

Este tutorial usa compute serverless e Unity Catalog. Para todas as opções de configuração que não forem especificadas, use as configurações default . Se compute serverless não estiver habilitada ou não for suportada no seu workspace, você poderá concluir o tutorial conforme escrito usando as configurações compute default .

Para criar um novo pipeline, siga estes passos:

1. Na sua workspace, clique em Na barra lateral, selecione "Novo" e, em seguida, "PipelineETL . Isso abre o editor pipeline com um nome pipeline default como New Pipeline <date> <time>.
2. (Opcional) Selecione o nome e insira um nome descritivo para o pipeline.
3. (Opcional) À direita do nome, clique no catálogo e no esquema para definir valores padrão diferentes.
4. (Opcional) No arquivo de origem my_transformation criado para você, selecione Python ou SQL na lista suspensa de idiomas para definir o idioma do arquivo.
5. Clique Use o código de exemplo .

O código de exemplo no idioma selecionado aparece no arquivo my_transformation na pasta transformations .

o passo 2: Desenvolva a lógica do seu pipeline

Nesta etapa, você usará o EditorLakeFlow Pipelines para desenvolver e validar o código-fonte do pipeline de forma interativa.

O código usa Auto Loader para ingestão incremental de dados. O Auto Loader detecta e processa automaticamente novos arquivos à medida que eles chegam ao armazenamento de objetos na nuvem. Para saber mais, consulte O que é o Auto Loader?

Um arquivo de código-fonte de exemplo foi criado na pasta de transformações do seu pipeline. Por default, todos os arquivos *.py e *.sql na pasta transformações fazem parte do código-fonte do seu pipeline.

1. Na pasta transformações, abra o arquivo de transformações de exemplo e substitua o código de exemplo pelo seguinte. Certifique-se de usar o idioma que você selecionou na etapa 1.

Python

Python

# Import modules
from pyspark import pipelines as dp
from pyspark.sql.functions import *
from pyspark.sql.types import DoubleType, IntegerType, StringType, StructType, StructField

# Define the path to the source data
file_path = f"/databricks-datasets/songs/data-001/"

# Define a streaming table to ingest data from a volume
schema = StructType(
  [
    StructField("artist_id", StringType(), True),
    StructField("artist_lat", DoubleType(), True),
    StructField("artist_long", DoubleType(), True),
    StructField("artist_location", StringType(), True),
    StructField("artist_name", StringType(), True),
    StructField("duration", DoubleType(), True),
    StructField("end_of_fade_in", DoubleType(), True),
    StructField("key", IntegerType(), True),
    StructField("key_confidence", DoubleType(), True),
    StructField("loudness", DoubleType(), True),
    StructField("release", StringType(), True),
    StructField("song_hotnes", DoubleType(), True),
    StructField("song_id", StringType(), True),
    StructField("start_of_fade_out", DoubleType(), True),
    StructField("tempo", DoubleType(), True),
    StructField("time_signature", DoubleType(), True),
    StructField("time_signature_confidence", DoubleType(), True),
    StructField("title", StringType(), True),
    StructField("year", IntegerType(), True),
    StructField("partial_sequence", IntegerType(), True)
  ]
)

@dp.table(
  comment="Raw data from a subset of the Million Song Dataset; a collection of features and metadata for contemporary music tracks."
)
def songs_raw():
  return (spark.readStream
    .format("cloudFiles")
    .schema(schema)
    .option("cloudFiles.format", "csv")
    .option("sep","\t")
    .load(file_path))

# Define a materialized view that validates data and renames a column
@dp.materialized_view(
  comment="Million Song Dataset with data cleaned and prepared for analysis."
)
@dp.expect("valid_artist_name", "artist_name IS NOT NULL")
@dp.expect("valid_title", "song_title IS NOT NULL")
@dp.expect("valid_duration", "duration > 0")
def songs_prepared():
  return (
    spark.read.table("songs_raw")
      .withColumnRenamed("title", "song_title")
      .select("artist_id", "artist_name", "duration", "release", "tempo", "time_signature", "song_title", "year")
  )

# Define a materialized view that has a filtered, aggregated, and sorted view of the data
@dp.materialized_view(
  comment="A table summarizing counts of songs released by the artists who released the most songs each year."
)
def top_artists_by_year():
  return (
    spark.read.table("songs_prepared")
      .filter(expr("year > 0"))
      .groupBy("artist_name", "year")
      .count().withColumnRenamed("count", "total_number_of_songs")
      .sort(desc("total_number_of_songs"), desc("year"))
  )

SQL

SQL

-- Define a streaming table to ingest data from a volume
CREATE OR REFRESH STREAMING TABLE songs_raw
COMMENT "Raw data from a subset of the Million Song Dataset; a collection of features and metadata for contemporary music tracks."
AS SELECT *
FROM STREAM read_files(
  '/databricks-datasets/songs/data-001/part*',
  format => "csv",
  header => "false",
  delimiter => "\t",
  schema => """
    artist_id STRING,
    artist_lat DOUBLE,
    artist_long DOUBLE,
    artist_location STRING,
    artist_name STRING,
    duration DOUBLE,
    end_of_fade_in DOUBLE,
    key INT,
    key_confidence DOUBLE,
    loudness DOUBLE,
    release STRING,
    song_hotnes DOUBLE,
    song_id STRING,
    start_of_fade_out DOUBLE,
    tempo DOUBLE,
    time_signature INT,
    time_signature_confidence DOUBLE,
    title STRING,
    year INT,
    partial_sequence STRING
  """,
  schemaEvolutionMode => "none");

-- Define a materialized view that validates data and renames a column
CREATE OR REFRESH MATERIALIZED VIEW songs_prepared(
CONSTRAINT valid_artist_name EXPECT (artist_name IS NOT NULL),
CONSTRAINT valid_title EXPECT (song_title IS NOT NULL),
CONSTRAINT valid_duration EXPECT (duration > 0)
)
COMMENT "Million Song Dataset with data cleaned and prepared for analysis."
AS SELECT artist_id, artist_name, duration, release, tempo, time_signature, title AS song_title, year
FROM songs_raw;

-- Define a materialized view that has a filtered, aggregated, and sorted view of the data
CREATE OR REFRESH MATERIALIZED VIEW top_artists_by_year
COMMENT "A table summarizing counts of songs released by the artists each year, who released the most songs."
AS SELECT
  artist_name,
  year,
  COUNT(*) AS total_number_of_songs
FROM songs_prepared
WHERE year > 0
GROUP BY artist_name, year
ORDER BY total_number_of_songs DESC, year DESC;

Esta fonte inclui código para três consultas. Você também pode colocar essas consultas em arquivos separados para organizar os arquivos e codificar da maneira que preferir.

2. Clique arquivo de execução ou pipelinede execução para iniciar uma atualização para o pipeline conectado. Com apenas um arquivo de origem no seu pipeline, eles são funcionalmente equivalentes.

Quando a atualização for concluída, o editor será atualizado com informações sobre seu pipeline.

• O gráfico de pipeline, também chamado de gráfico acíclico dirigido (DAG), na barra lateral à direita do seu código, mostra três tabelas, songs_raw, songs_prepared e top_artists_by_year.
• Um resumo da atualização é exibido na parte superior do navegador ativo pipeline .
• Os detalhes das tabelas geradas são mostrados no painel inferior, e você pode navegar pelos dados das tabelas selecionando uma.

Isso inclui dados brutos e limpos, bem como algumas análises simples para encontrar os principais artistas por ano. No próximo passo, você cria consultas ad-hoc para análise posterior em um arquivo separado no seu pipeline.

o passo 3: Explore o conjunto de dados criado pelo seu pipeline

Nesta etapa, você executa consultas ad-hoc nos dados processados no pipeline ETL para analisar os dados da música no Editor Databricks SQL . Essas consultas usam os registros preparados criados no passo anterior.

Primeiro, execute uma consulta que encontre os artistas que lançaram o maior número de músicas por ano desde 1990.

1. Na barra lateral do navegador ativo pipeline , clique em Adicione então a Exploração .

2. Digite um nome e selecione SQL para o arquivo de exploração. Um SQL Notebook é criado em uma nova pasta explorations . Os arquivos na pasta explorations não são executados como parte de uma atualização pipeline por default. O SQL Notebook tem células que você pode executar juntas ou separadamente.

3. Para criar uma tabela de artistas que lançaram mais músicas em cada ano após 1990, insira o seguinte código no novo arquivo SQL (se houver código de exemplo no arquivo, substitua-o). Como este Notebook não faz parte do pipeline, ele não usa o catálogo e o esquema default . Substitua o <catalog>.<schema> pelo catálogo e esquema que você usou como padrão para o pipeline:

   SQL

   -- Which artists released the most songs each year in 1990 or later?
   SELECT artist_name, total_number_of_songs, year
     -- replace with the catalog/schema you are using:
     FROM <catalog>.<schema>.top_artists_by_year
     WHERE year >= 1990
     ORDER BY total_number_of_songs DESC, year DESC;

4. Clique ou pressione Shift + Enter para executar esta consulta.

Agora, execute outra consulta que encontre músicas com uma batida 4/4 e ritmo dançante.

1. Adicione o seguinte código à próxima célula no mesmo arquivo. Novamente, substitua o <catalog>.<schema> pelo catálogo e esquema que você usou como padrão para o pipeline:

   SQL

   -- Find songs with a 4/4 beat and danceable tempo
   SELECT artist_name, song_title, tempo
     -- replace with the catalog/schema you are using:
     FROM <catalog>.<schema>.songs_prepared
     WHERE time_signature = 4 AND tempo between 100 and 140;

2. Clique ou pressione Shift + Enter para executar esta consulta.

Etapa 4: Criar um trabalho para executar o pipeline

A seguir, crie um fluxo de trabalho para automatizar a ingestão de dados, processamento e análise dos passos usando um Databricks Job que é executado em um programar.

1. Na parte superior do editor, escolha o botão programar .
2. Se a caixa de diálogo do programa aparecer, escolha Adicionar programa .
3. Isso abre a caixa de diálogo Novo programa , onde você pode criar um Job para executar seu pipeline em um programa.
4. Opcionalmente, dê um nome ao trabalho.
5. Por default, o programador é configurado para executar uma vez por dia. Você pode aceitar esse padrão ou definir seu próprio programa. Escolher Avançado lhe dá a opção de definir um horário específico para a execução do trabalho. Selecionar Mais opções permite que você crie notificações quando o trabalho for executado.
6. Selecione Criar para aplicar as alterações e criar o trabalho.

Agora o Job será executado diariamente para manter seu pipeline atualizado. Você pode escolher programar novamente para view a lista de programas. Você pode gerenciar programas para seu pipeline a partir dessa caixa de diálogo, incluindo adicionar, editar ou remover programas.

Clicar no nome do programa (ou Job) leva você para a página do Job na lista Jobs & pipeline . A partir daí você pode view detalhes sobre a execução do Job, incluindo a história de execução, ou executar o Job imediatamente com o botão executar agora .

Consulte Monitorar Jobs do LakeFlow para mais informações sobre execuções de jobs.

Recursos adicionais

• Para saber mais sobre pipelines de processamento de dados, consulte Spark Declarative Pipelines no Lakeflow
• Para saber mais sobre o Databricks Notebook, consulte Databricks Notebook.
• Para saber mais sobre LakeFlow Jobs, consulte What are Job?
• Para saber mais sobre o Delta Lake, consulte O que é o Delta Lake na Databricks?