Executar sua primeira carga de trabalho de ETL no Azure Databricks

Saiba como usar as ferramentas prontas para produção do Azure Databricks para desenvolver e implantar seus primeiros pipelines de ETL (extração, transformação e carregamento) para orquestração de dados.

Ao final deste artigo, você se sentirá confortável:

1. Como iniciar um cluster de computação para todas as finalidades do Databricks.
2. Criar um notebook do Databricks.
3. Como configurar a ingestão incremental de dados para o Delta Lake com o Carregador Automático.
4. Executar células de notebook para processar, consultar e visualizar dados.
5. Agendar um notebook como um trabalho do Databricks.

Este tutorial usa notebooks interativos para concluir tarefas de ETL comuns no Python ou no Scala.

Use também o Delta Live Tables para criar pipelines de ETL. O Databricks criou o Delta Live Tables para reduzir a complexidade da criação, da implantação e da manutenção de pipelines de ETL de produção. Consulte Tutorial: como executar seu primeiro pipeline do Delta Live Tables.

Use também o provedor Terraform do Databricks para criar os recursos deste artigo. Confira Criar clusters, notebooks e trabalhos com o Terraform.

Requisitos

• Você está conectado a um workspace do Azure Databricks.
• Você tem permissão para criar um cluster.

Observação

Se você não tiver privilégios de controle de cluster, ainda poderá concluir a maioria das etapas abaixo, desde que tenha acesso a um cluster.

Etapa 1: criar um cluster

Para fazer a análise de dados exploratória e a engenharia de dados, crie um cluster para fornecer os recursos de computação necessários para executar comandos.

1. Clique em Computação na barra lateral.
2. Na página Computação, clique em Criar Cluster. Isso abre a página Novo Cluster.
3. Especifique um nome exclusivo para o cluster, mantenha os valores restantes no estado padrão e clique em Criar Cluster.

Para saber mais sobre clusters do Databricks, confira Computação.

Etapa 2: criar um notebook do Databricks

Para começar a gravar e executar código interativo no Azure Databricks, crie um notebook.

1. Clique no Novo na barra lateral e clique em Notebook.
2. Na página Criar Notebook:
   • Especifique um nome exclusivo para seu notebook.
   • Verifique se a linguagem padrão está definida como Python ou Scala.
   • Selecione o cluster que você criou na etapa 1 na lista suspensa cluster.
   • Clique em Criar.

Um notebook é aberto com uma célula vazia na parte superior.

Para saber mais sobre como criar e gerenciar notebooks, consulte Gerenciar notebooks.

Etapa 3: Configurar o Carregador Automático para ingerir dados no Delta Lake

O Databricks recomenda o uso do Carregador Automático para ingestão de dados incremental. O Carregador Automático detecta e processa automaticamente arquivos novos confirme eles chegam no armazenamento de objeto da nuvem.

O Databricks recomenda armazenar dados com o Delta Lake. O Delta Lake é uma camada de armazenamento código aberto que fornece transações ACID e habilita o data lakehouse. O Delta Lake é o formato padrão para tabelas criadas no Databricks.

A fim de configurar o Carregador Automático para ingerir dados em uma tabela do Delta Lake, copie e cole o seguinte código em uma célula vazia no notebook:

Python

# Import functions
from pyspark.sql.functions import col, current_timestamp

# Define variables used in code below
file_path = "/databricks-datasets/structured-streaming/events"
username = spark.sql("SELECT regexp_replace(current_user(), '[^a-zA-Z0-9]', '_')").first()[0]
table_name = f"{username}_etl_quickstart"
checkpoint_path = f"/tmp/{username}/_checkpoint/etl_quickstart"

# Clear out data from previous demo execution
spark.sql(f"DROP TABLE IF EXISTS {table_name}")
dbutils.fs.rm(checkpoint_path, True)

# Configure Auto Loader to ingest JSON data to a Delta table
(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load(file_path)
  .select("*", col("_metadata.file_path").alias("source_file"), current_timestamp().alias("processing_time"))
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable(table_name))

Scala

// Imports
import org.apache.spark.sql.functions.current_timestamp
import org.apache.spark.sql.streaming.Trigger
import spark.implicits._

// Define variables used in code below
val file_path = "/databricks-datasets/structured-streaming/events"
val username = spark.sql("SELECT regexp_replace(current_user(), '[^a-zA-Z0-9]', '_')").first.get(0)
val table_name = s"${username}_etl_quickstart"
val checkpoint_path = s"/tmp/${username}/_checkpoint"

// Clear out data from previous demo execution
spark.sql(s"DROP TABLE IF EXISTS ${table_name}")
dbutils.fs.rm(checkpoint_path, true)

// Configure Auto Loader to ingest JSON data to a Delta table
spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load(file_path)
  .select($"*", $"_metadata.file_path".as("source_file"), current_timestamp.as("processing_time"))
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(Trigger.AvailableNow)
  .toTable(table_name)

Observação

As variáveis definidas nesse código devem permitir que você o execute com segurança sem risco de conflito com ativos de workspace existentes ou outros usuários. Permissões restritas de rede ou armazenamento gerarão erros ao executar esse código, por isso contate o administrador do workspace para solucionar problemas dessas restrições.

Para saber mais sobre o Carregador Automático, confira O que é o Carregador Automático?.

Etapa 4: Processar e interagir com dados

Os notebooks executam célula por célula lógica. Para executar a lógica na célula:

1. Para executar a célula concluída na etapa anterior, selecione a célula e pressione SHIFT+ENTER.

2. Para consultar a tabela que você acabou de criar, copie e cole o código a seguir em uma célula vazia e pressione SHIFT+ENTER para executar a célula.

   Python

   df = spark.read.table(table_name)
   

   Scala

   val df = spark.read.table(table_name)
   

3. Para visualizar os dados em seu DataFrame, copie e cole o código a seguir em uma célula vazia e pressione SHIFT+ENTER para executar a célula.

   Python

   display(df)
   

   Scala

   display(df)
   

Para saber mais sobre opções interativas para visualizar dados, consulte Visualizações em notebooks do Databricks.

Etapa 5: Agendar um trabalho

Você pode executar notebooks do Databricks como scripts de produção adicionando-os como uma tarefa em um trabalho do Databricks. Nesta etapa, você criará um novo trabalho que poderá disparar manualmente.

Para agendar seu notebook como uma tarefa:

1. Clique em Agendamento no lado direito da barra de cabeçalho.
2. Insira um nome exclusivo para o nome do trabalho.
3. Clique em Manual.
4. Na lista suspensa cluster, selecione o cluster que você criou na etapa 1.
5. Clique em Criar.
6. Na janela exibida, clique em Executar agora.
7. Para consultar os resultados de execução de trabalho, clique no ícone próximo ao timestamp de Última execução.

Para obter mais informações sobre trabalhos, consulte O que são Trabalhos do Azure Databricks?.

Integrações adicionais

Saiba mais sobre integrações e ferramentas para engenharia de dados com o Azure Databricks:

• Conectar seu IDE favorito
• Usar o dbt com o Databricks
• Saiba sobre a Interface de Linha de Comando (CLI) do Databricks
• Saiba mais sobre o Provedor Do Databricks Terraform