Como criar uma Delta Live Table em Lakehouse: passo a passo

Este tutorial mostra como criar uma Delta Live Table em Lakehouse para ingestão contínua e transformação de dados, útil quando necessitas de pipelines fiáveis com historial e monitorização. A tarefa foca um exemplo concreto: ler ficheiros JSON de um diretório, aplicar transformações simples e publicar uma tabela Delta pronta para consumo.

Pré-requisitos

• Conta com acesso ao Microsoft Fabric e permissões para criar um Lakehouse e pipelines.
• Um Lakehouse com um diretório para ficheiros de ingestão (ex.: /lakehouse/raw/events/).
• Noções básicas de Python/PySpark e do formato Delta.
• Exemplo de ficheiros JSON com esquema consistente.

Passo 1: Entender o que é uma Delta Live Table em Lakehouse

Delta Live Table é uma abordagem para construir pipelines declarativos que produzem tabelas Delta geridas. Permite ingestão contínua, tratamento de esquemas e monitorização integrada. Este passo é conceptual: visualiza o fluxo — origem (ficheiros JSON) → transformação (limpeza, tipos) → destino (tabela Delta).

Passo 2: Criar um workspace de pipeline

Criar um pipeline no ambiente do Fabric que execute código PySpark em trigger contínuo ou agendado. Define o nome, o tipo de runtime e o Lakehouse onde as tabelas serão gravadas.

# Exemplo conceptual (interface do Fabric tem gui); se usares CLI/SDK configura o pipeline com:
# runtime: pyspark
# target: Lakehouse//tables/

Passo 3: Código PySpark mínimo para uma Delta Live Table

Escreve um script PySpark que lê JSON em modo de streaming, aplica transformações simples e escreve para uma tabela Delta. Mantém o código mínimo para testar o fluxo.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_timestamp

spark = SparkSession.builder.getOrCreate()

# Fonte: ficheiros JSON em modo 'cloudFiles' (auto-infer schema quando disponível)
raw_df = (
    spark.readStream
         .format("cloudFiles")
         .option("cloudFiles.format", "json")
         .option("cloudFiles.inferColumnTypes", "true")
         .load("/lakehouse/raw/events/")
)

# Transformações: limpar campos e converter timestamps
clean_df = (
    raw_df
    .withColumn("event_time", to_timestamp(col("event_time"), "yyyy-MM-dd'T'HH:mm:ss"))
    .withColumn("user_id", col("user_id").cast("string"))
    .na.drop(subset=["event_time", "user_id"])  # descartar linhas inválidas
)

# Escrita em Delta para o Lakehouse (modo append, checkpoint obrigatório)
(output = clean_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/lakehouse/checkpoints/events_checkpoint/")
    .start("/lakehouse/tables/events_delta/"))

Passo 4: Configurar políticas de esquema e tratamento de erros

Define como o pipeline lida com alterações de esquema e com dados corruptos. Em ambientes Fabric/Lakehouse, configura opções de Auto Loader ou cloudFiles para ignorar ficheiros corruptos e registar erros.

# Opções adicionais para robustez (exemplo cloudFiles)
.option("cloudFiles.schemaEvolutionMode", "addNewColumns")
.option("cloudFiles.maxFilesPerTrigger", "100")
.option("badRecordsPath", "/lakehouse/errors/bad_records/")

Passo 5: Validar e criar a tabela Delta gerida

Depois da stream escrita para o caminho Delta, regista a pasta como tabela Delta no catálogo do Lakehouse para consulta SQL e consumo por Power BI.

-- SQL executado no SQL endpoint ou notebook para criar tabela a partir do caminho Delta
CREATE TABLE IF NOT EXISTS lakehouse.events_delta
USING DELTA
LOCATION '/lakehouse/tables/events_delta/';

Passo 6: Operações comuns e erros a evitar

Monitora checkpointLocation para evitar perda de estado; não partilhes o mesmo checkpoint entre pipelines distintos. Se encontrares erro de schema mismatch, activa schemaEvolution ou faz validação prévia. Evita escrever diretamente para o mesmo diretório com jobs concorrentes sem ACID (Delta resolve parcialmente, mas tem limites).

# Erros comuns (consulta de diagnóstico)
-- Verifica o estado da tabela
DESCRIBE HISTORY lakehouse.events_delta;
-- Conferir ficheiros corruptos
ls /lakehouse/errors/bad_records/

Verificar o resultado

Confirma que a tabela Delta recebeu dados: consulta a tabela com SQL e verifica que há dados recentes. Verifica a existência do checkpoint e os ficheiros _delta_log no diretório da tabela.

SELECT COUNT(*) FROM lakehouse.events_delta;
SELECT event_time, user_id FROM lakehouse.events_delta ORDER BY event_time DESC LIMIT 10;
# No storage, confirma:
# /lakehouse/tables/events_delta/_delta_log/
# /lakehouse/checkpoints/events_checkpoint/

Conclusão

Agora tens um pipeline básico de Delta Live Table em Lakehouse para ingestão contínua de JSON, transformação mínima e publicação como tabela Delta. Próximos passos: acrescentar testes de qualidade, enriquecimentos e monitorização no Fabric. Dica: começa com um conjunto pequeno de ficheiros para validar o esquema antes de ativar o streaming em produção.