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 o Databricks Notebook e um Databricks Job para orquestrar um fluxo de trabalho, o Databricks recomenda o uso do Delta Live TablesO site recomenda o uso do Notebook, uma interface declarativa para criar um pipeline de processamento de dados confiável, sustentável e testável.
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
Você está conectado ao Databricks e ao workspace Data Science & Engineering.
O senhor tem permissão para criar clusters ou acessar clusters.
(Opcional) Para publicar tabelas no Unity Catalog, você deve criar um catálogo e um esquema no Unity Catalog.
Exemplo: Conjunto de dados de milhões de músicas
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 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.
Clique em Calcular na barra lateral.
Na página Computação, clique em Criar cluster.
Na página Novo cluster, insira um nome exclusivo para o cluster.
No Modo de acesso, selecione Usuário único.
Em Acesso único de usuário ou serviço principal, selecione seu nome de usuário.
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.
Na barra lateral, clique em Novo e selecione Notebook no menu. A caixa de diálogo Criar Notebook é exibida.
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.
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
.Clique 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 emfile_path
e grava os dados na tabela especificada portable_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.
Na barra lateral, clique em Novo e selecione Notebook no menu. A caixa de diálogo Criar Notebook é exibida.
Insira um nome para o notebook. Por exemplo,
Prepare songs data
. Altere o idioma padrão para SQL.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.Clique , 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.
Na barra lateral, clique em Novo e selecione Notebook no menu. A caixa de diálogo Criar Notebook é exibida.
Insira um nome para o notebook. Por exemplo,
Analyze songs data
. Altere o idioma padrão para SQL.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
.Clique 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
.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.
Em seu workspace do Data Science & Engineering, faça um dos seguintes:
Clique em fluxo de trabalho na barra lateral e clique em .
Na barra lateral, clique em Novo e selecione Job.
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".
Em Nome da tarefa, insira um nome para a primeira tarefa, por exemplo,
Ingest_songs_data
.Em Tipo, selecione o tipo de tarefa Notebook.
Em Origem, selecione workspace.
Use o navegador de arquivos para localizar o notebook de ingestão de dados, clique no nome do notebook e clique em Confirmar.
Em Cluster, selecione Shared_job_cluster ou o cluster que você criou na etapa
Create a cluster
.Clique em Create.
Clique abaixo da tarefa que você acabou de criar e selecione Notebook.
Em Nome da tarefa, insira um nome para a tarefa, por exemplo,
Prepare_songs_data
.Em Tipo, selecione o tipo de tarefa Notebook.
Em Origem, selecione workspace.
Use o navegador de arquivos para localizar o notebook de preparação de dados, clique no nome do notebook e clique em Confirmar.
Em Cluster, selecione Shared_job_cluster ou o cluster que você criou na etapa
Create a cluster
.Clique em Create.
Clique abaixo da tarefa que você acabou de criar e selecione Notebook.
Em Nome da tarefa, insira um nome para a tarefa, por exemplo,
Analyze_songs_data
.Em Tipo, selecione o tipo de tarefa Notebook.
Em Origem, selecione workspace.
Use o navegador de arquivos para localizar o notebook de análise de dados, clique no nome do notebook e clique em Confirmar.
Em Cluster, selecione Shared_job_cluster ou o cluster que você criou na etapa
Create a cluster
.Clique em Create.
Para executar o fluxo de trabalho, clique em . Para view os detalhes da execução, clique no link na coluna de tempo de início da execução no Job execução view. Clique em cada tarefa para acessar view detalhes da execução da tarefa.
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:
Clique em fluxo de trabalho na barra lateral.
Na coluna Nome, clique no nome do trabalho. O painel lateral exibe os detalhes do trabalho.
Clique em Adicionar acionador no painel Detalhes do trabalho e selecione Programado no Tipo de acionador.
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.
Clique em Salvar.
Saiba mais
Para saber mais sobre os notebooks do Databricks, consulte Introdução aos notebooks do Databricks.
Para saber mais sobre os empregos da Databricks, consulte O que são empregos da Databricks?
Para saber mais sobre o Delta Lake, consulte O que é o Delta Lake?.
Para saber mais sobre pipelines de processamento de dados com o Delta Live Tables, consulte O que é o Delta Live Tables?.