Crie um pipeline de dados completo no Databricks

Este artigo demonstra como criar e implementar um pipeline de processamento de dados de ponta a ponta, abrangendo desde a ingestão de dados brutos, a transformação dos dados e a realização de análises nos dados processados.

Observação

Embora este artigo demonstre como criar um pipeline de dados completo usando Databricks Notebook e um Databricks Job para orquestrar um fluxo de trabalho, Databricks recomenda usar Delta Live Tables, uma interface declarativa para criar pipelines de processamento de dados confiáveis, sustentáveis e testáveis.

O que é um pipeline de dados?

Um pipeline de dados implementa as etapas necessárias para mover dados dos sistemas de origem, transformar esses dados com base nos requisitos e armazenar os dados em um sistema de destino. Um pipeline de dados inclui todos os processos necessários para transformar dados brutos em dados preparados que os usuários podem consumir. Por exemplo, um pipeline de dados pode preparar os dados para que os analistas de dados e cientistas de dados possam extrair valor dos dados por meio de análises e relatórios.

Um fluxo de trabalho de extração, transformação e carregamento (ETL) é um exemplo comum de um pipeline de dados. No processamento ETL, os dados são coletados dos sistemas de origem e armazenados em uma área de estágio, são transformados conforme os requisitos (garantindo qualidade dos dados, eliminando registros duplicados, entre outros) e, em seguida, são gravados em um sistema de destino, como um data warehouse ou um data lake.

Etapas do pipeline de dados

Para ajudá-lo a começar a criar pipelines de dados no Databricks, o exemplo incluído neste artigo mostra a criação de um fluxo de trabalho de processamento de dados:

  • Use os recursos do Databricks para explorar um conjunto de dados brutos.

  • Crie um notebook Databricks para fazer a ingestão de dados de origem brutos e escrever os dados brutos em uma tabela de destino.

  • Crie um notebook para transformar os dados de origem brutos e escreva os dados transformados em uma tabela de destino.

  • Crie um notebook Databricks para consultar os dados transformados.

  • Automatize o pipeline de dados com um trabalho do Databricks.

Requisitos

Exemplo: Conjunto de dados de milhões de músicas

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

Etapa 1: criar um cluster

Para realizar o processamento e análise de dados neste exemplo, crie um cluster para fornecer os recursos de computação necessários para executar comandos.

Observação

Como este exemplo usa um exemplo de dataset armazenado no DBFS e recomenda a persistência de tabelas no Unity Catalog, o senhor cria um cluster configurado com o modo de acesso de usuário único. O modo de acesso de usuário único fornece acesso total ao DBFS e também permite o acesso ao Unity Catalog. Consulte Práticas recomendadas para DBFS e Unity Catalog.

  1. Clique em Calcular na barra lateral.

  2. Na página Computação, clique em Criar cluster.

  3. Na página Novo cluster, insira um nome exclusivo para o cluster.

  4. No Modo de acesso, selecione Usuário único.

  5. Em Acesso único de usuário ou serviço principal, selecione seu nome de usuário.

  6. Deixe os valores restantes em seu estado padrão e clique em Criar cluster.

Para saber mais sobre clusters Databricks, consulte Compute.

Etapa 2: Explorar os dados de origem

Para saber como usar a interface Databricks para explorar os dados brutos de origem, consulte Explorar os dados de origem para um pipeline de dados. Se você quiser ir diretamente para a ingestão e preparação dos dados, vá para a passo 3: Ingerir os dados brutos.

Etapa 3: ingerir os dados brutos

Nesta etapa, você carrega os dados brutos em uma tabela para disponibilizá-los para processamento posterior. Para gerenciar dados ativos na plataforma Databricks como tabelas, a Databricks recomenda o Unity Catalog. No entanto, se você não tiver permissões para criar o catálogo e o esquema necessários para publicar tabelas no Unity Catalog, ainda poderá concluir as etapas a seguir publicando tabelas no Hive metastore.

Para ingerir dados, o Databricks recomenda usar o Auto Loader. O Auto Loader detecta e processa automaticamente novos arquivos à medida que eles chegam ao armazenamento de objetos na nuvem.

Você pode configurar o Auto Loader para detectar automaticamente o esquema dos dados carregados, permitindo que você inicialize tabelas sem declarar explicitamente o esquema dos dados e adapte o esquema da tabela à medida que novas colunas são introduzidas. Isso elimina a necessidade de rastrear manualmente e aplicar alterações de esquema ao longo do tempo. Databricks recomenda a inferência de esquema ao usar Auto Loader. No entanto, como visto na passo de exploração de dados, os dados das músicas não contêm informações de cabeçalho. Como o cabeçalho não é armazenado com os dados, você precisará definir explicitamente o esquema, conforme mostrado no próximo exemplo.

  1. Na barra lateral, clique em Novo ícone Novo e selecione Notebook no menu. A caixa de diálogo Criar Notebook é exibida.

  2. Insira um nome para o notebook, por exemplo, Ingest songs data. Por padrão:

    • Python é a linguagem selecionada.

    • O Notebook está anexado aos últimos clusters que você usou. Neste caso, os clusters que você criou no passo 1: Crie um clusters.

  3. Insira o seguinte na primeira célula do notebook:

    from pyspark.sql.types import DoubleType, IntegerType, StringType, StructType, StructField

# Define variables used in the code below
file_path = "/databricks-datasets/songs/data-001/"
table_name = "<table-name>"
checkpoint_path = "/tmp/pipeline_get_started/_checkpoint/song_data"

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)
  ]
)

(spark.readStream
  .format("cloudFiles")
  .schema(schema)
  .option("cloudFiles.format", "csv")
  .option("sep","\t")
  .load(file_path)
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable(table_name)
)

    Se você estiver usando o Unity Catalog, <table-name> substitua por um catálogo, esquema e nome de tabela para conter os registros ingeridos (por exemplo, data_pipelines.songs_data.raw_song_data). Caso contrário, <table-name> substitua pelo nome de uma tabela para conter os registros ingeridos, por exemplo, raw_song_data.

    Substitua <checkpoint-path> por um caminho para um diretório no DBFS para manter arquivos de ponto de verificação, por exemplo, /tmp/pipeline_get_started/_checkpoint/song_data.

  4. Clique Menu Executar e selecione Célula de execução. Este exemplo define o esquema de dados usando a informação do README, ingere os dados das músicas de todos os arquivos contidos em file_path e grava os dados na tabela especificada por table_name.

Etapa 4: Preparar os dados brutos

Para preparar os dados brutos para análise, os seguintes passos transformam os dados brutos das músicas filtrando as colunas não necessárias e adicionando um novo campo contendo um registro de data e hora para a criação do novo registro.

  1. Na barra lateral, clique em Novo ícone Novo e selecione Notebook no menu. A caixa de diálogo Criar Notebook é exibida.

  2. Insira um nome para o notebook. Por exemplo, Prepare songs data. Altere o idioma padrão para SQL.

  3. Insira o seguinte na primeira célula do notebook:

    CREATE OR REPLACE TABLE
  <table-name> (
    artist_id STRING,
    artist_name STRING,
    duration DOUBLE,
    release STRING,
    tempo DOUBLE,
    time_signature DOUBLE,
    title STRING,
    year DOUBLE,
    processed_time TIMESTAMP
  );

INSERT INTO
  <table-name>
SELECT
  artist_id,
  artist_name,
  duration,
  release,
  tempo,
  time_signature,
  title,
  year,
  current_timestamp()
FROM
  <raw-songs-table-name>

    Se você estiver utilizando o Unity Catalog, substitua <table-name> por um catálogo, esquema e nome da tabela para conter os registros filtrados e transformados (por exemplo, data_pipelines.songs_data.prepared_song_data). Caso contrário, substitua <table-name> pelo nome de uma tabela para conter os registros filtrados e transformados (por exemplo, prepared_song_data).

    Substitua <raw-songs-table-name> pelo nome da tabela que contém os registros de músicas brutas ingeridas na passo anterior.

  4. Clique Menu Executar, e selecione a célula de execução.

Etapa 5: Consultar os dados transformados

Nesta etapa, você estende o pipeline de processamento adicionando uma consulta para analisar os dados das músicas. Essas consultas usam os registros preparados criados na passo anterior.

  1. Na barra lateral, clique em Novo ícone Novo e selecione Notebook no menu. A caixa de diálogo Criar Notebook é exibida.

  2. Insira um nome para o notebook. Por exemplo, Analyze songs data. Altere o idioma padrão para SQL.

  3. Insira o seguinte na primeira célula do notebook:

    -- Which artists released the most songs each year?
SELECT
  artist_name,
  count(artist_name)
AS
  num_songs,
  year
FROM
  <prepared-songs-table-name>
WHERE
  year > 0
GROUP BY
  artist_name,
  year
ORDER BY
  num_songs DESC,
  year DESC

    Substitua <prepared-songs-table-name> pelo nome da tabela que contém os dados preparados. Por exemplo, data_pipelines.songs_data.prepared_song_data.

  4. Clique Abaixo do cursor no menu de ações da célula, selecione Adicionar célula abaixo e digite o seguinte na nova célula:

     -- Find songs for your DJ list
 SELECT
   artist_name,
   title,
   tempo
 FROM
   <prepared-songs-table-name>
 WHERE
   time_signature = 4
   AND
   tempo between 100 and 140;

    Substitua <prepared-songs-table-name> pelo nome da tabela preparada criada na passo anterior. Por exemplo, data_pipelines.songs_data.prepared_song_data.

  5. Para executar as consultas e visualizar a saída, clique em Executar tudo.

Etapa 6: criar um trabalho do Databricks para executar o pipeline

É possível criar um fluxo de trabalho para automatizar a execução das etapas de ingestão, processamento e análise de dados usando um trabalho Databricks.

  1. Em seu workspace do Data Science & Engineering, faça um dos seguintes:

    • Clique Ícone de jobs fluxo de trabalho na barra lateral e clique Botão criar job.

    • Na barra lateral, clique em Novo ícone Novo e selecione Job.

  2. Na caixa de diálogo da tarefa na guia Tarefas, substitua Adicionar um nome para o seu trabalho… pelo nome do seu trabalho. Por exemplo, "Fluxo de trabalho de músicas".

  3. Em Nome da tarefa, insira um nome para a primeira tarefa, por exemplo, Ingest_songs_data.

  4. Em Tipo, selecione o tipo de tarefa Notebook.

  5. Em Origem, selecione workspace.

  6. Use o navegador de arquivos para localizar o notebook de ingestão de dados, clique no nome do notebook e clique em Confirmar.

  7. Em Cluster, selecione Shared_job_cluster ou o cluster que você criou na etapa Create a cluster.

  8. Clique em Create.

  9. Clique Botão adicionar tarefa abaixo da tarefa que você acabou de criar e selecione Notebook.

  10. Em Nome da tarefa, insira um nome para a tarefa, por exemplo, Prepare_songs_data.

  11. Em Tipo, selecione o tipo de tarefa Notebook.

  12. Em Origem, selecione workspace.

  13. Use o navegador de arquivos para localizar o notebook de preparação de dados, clique no nome do notebook e clique em Confirmar.

  14. Em Cluster, selecione Shared_job_cluster ou o cluster que você criou na etapa Create a cluster.

  15. Clique em Create.

  16. Clique Botão adicionar tarefa abaixo da tarefa que você acabou de criar e selecione Notebook.

  17. Em Nome da tarefa, insira um nome para a tarefa, por exemplo, Analyze_songs_data.

  18. Em Tipo, selecione o tipo de tarefa Notebook.

  19. Em Origem, selecione workspace.

  20. Use o navegador de arquivos para localizar o notebook de análise de dados, clique no nome do notebook e clique em Confirmar.

  21. Em Cluster, selecione Shared_job_cluster ou o cluster que você criou na etapa Create a cluster.

  22. Clique em Create.

  23. Para executar o fluxo de trabalho, clique em Botão executar agora. Para ver os detalhes da execução, clique no link na coluna de horário de início da execução na visualização Job runs . Clique em cada tarefa para view os detalhes da execução da tarefa.

  24. Para visualizar os resultados quando o fluxo de trabalho for concluído, clique na tarefa final de análise de dados. A página Saída aparece e exibe os resultados da consulta.

Etapa 7: agendar o trabalho de pipeline de dados

Observação

Para demonstrar o uso de um Databricks Job para orquestrar um fluxo de trabalho agendado, este exemplo de começar separa as passos de ingestão, preparação e análise em Notebook separado, e cada Notebook é então usado para criar uma tarefa no Job. Se todo o processamento estiver contido em um único Notebook, você poderá programar facilmente o Notebook diretamente da interface do usuário do Databricks Notebook . Consulte Criar e gerenciar trabalhos agendados do notebook.

Um requisito comum é executar um pipeline de dados de forma programada. Para definir um agendamento para o trabalho que executa o pipeline:

  1. Clique Ícone de jobs fluxo de trabalho na barra lateral.

  2. Na coluna Nome, clique no nome do trabalho. O painel lateral exibe os detalhes do trabalho.

  3. Clique em Adicionar acionador no painel Detalhes do trabalho e selecione Programado no Tipo de acionador.

  4. Especifique o período, a hora inicial e o fuso horário. Opcionalmente, marque a caixa de seleção Mostrar Cron Syntax para exibir e editar a programação na Sintaxe Quartz Cron.

  5. Clique em Salvar.

Saiba mais