Teste de unidade para pipelines

Para obter informações gerais sobre teste de unidade em Python no Databricks, consulte teste de unidade em Python.

Os pipelines do Lakeflow suportam a escrita de testes unitários em Python no Editor web do Lakeflow Pipelines. Isso permite validar a lógica de transformação Python ou SQL usando dados fictícios. Com o framework de testes de pipeline, você pode testar casos de borda, validar APIs proprietárias de pipeline (Auto CDC, tabelas de streaming, expectativas, fluxos de anexação) e realizar iterações utilizando entradas simuladas (mocks) para operações com identificadores de tabela suportadas. Examine as limitações de isolamento antes de executar testes.

• Execução de teste isolada: a estrutura fornece uma SparkSession que redireciona operações de tabela para um esquema de teste temporário no catálogo padrão do pipeline, para que você possa simular dados de entrada e gravar saídas de teste sem afetar as tabelas de produção. O isolamento se aplica a operações que fazem referência a uma tabela por nome; consulte Limitações.
• Escopo de teste flexível: execute um subconjunto de um pipeline (tabelas individuais, cadeias de tabelas dependentes ou pipelines completos) na infraestrutura de computação do pipeline, utilizando a SparkSession de teste.
• Validação dos resultados: Verifique os resultados de tabelas de saída isoladas geradas em um teste usando asserções padrão do pytest.

Quando usar o teste de unidade

Os casos de uso típicos incluem:

• Validando uma nova lógica de transformação: teste se sua transformação produz o esquema esperado, contagens de linhas, agregações e lógica de negócios antes de executar em dados de produção.
• Testando especificações de CDC Automático: valide se as definições de fluxo do CDC Automático processam corretamente eventos de alteração, manipulando inserções, atualizações, exclusões e tipos SCD (Dimensão de Alteração Lenta), usando dados fictícios.
• Testando expectativas e regras de qualidade de dados: verifique se as expectativas falham quando deveriam e passam quando os dados são válidos.
• Testes em tabelas dependentes: Teste cadeias de transformações (por exemplo, bronze, prata e ouro) para validar se os dados fluem corretamente por meio do grafo do pipeline.

Requirements

• Permissão Owner de pipeline, além dos privilégios USE CATALOG e CREATE SCHEMA no catálogo padrão do pipeline. A estrutura precisa desses privilégios para criar o esquema de teste temporário em que os testes são executados.

  Para verificar ou configurar a permissão do pipeline, abra o pipeline e clique em Compartilhar. Você deve ser o pipeline Owner (IS OWNER); CAN RUN e CAN MANAGE não são suficientes para executar testes. Consulte Configurar permissões de pipeline.

  Para verificar ou definir os privilégios do catálogo, abra o catálogo no Catalog Explorer, selecione a guia Permissões e confirme se você tem USE CATALOG e CREATE SCHEMA. Um proprietário de catálogo, um administrador de metastore ou um usuário com o MANAGE privilégio podem concedê-los, inclusive com SQL:

  GRANT USE CATALOG, CREATE SCHEMA ON CATALOG <catalog_name> TO `<principal>`;
  

  Para obter mais informações, consulte Referência de privilégios do Unity Catalog.

• O pipeline deve ser configurado no modo acionado (não contínuo).

• O pipeline deve estar no canal PREVIEW. Os testes unitários estão em Beta e estão disponíveis apenas no PREVIEW.

• Não há suporte para o Spark Connect.

Limitações

O isolamento de teste é feito apenas pelo nome da tabela

• Não leia nem grave por caminho ou conector. O isolamento redireciona somente as operações que fazem referência a uma tabela por nome (por exemplo, spark.read.table("catalog.schema.table") ou df.write.saveAsTable("catalog.schema.table")). As operações tratadas por um caminho ou por meio de um conector ignoram o isolamento e atuam diretamente em sistemas de produção reais:

  • A gravação por caminho (por exemplo, df.write.save("/Volumes/..."), um caminho dbfs:/ ou um caminho de nuvem ou local externo, como s3://... ou abfss://...) grava no armazenamento de produção real e pode sobrescrever dados de produção.
  • A leitura por caminho (por exemplo, spark.read.load(path) ou spark.read.format("delta").load(path)) retorna dados reais de produção em vez de sua simulação.
  • A leitura a partir de um conector conecta-se à fonte de produção real. Isso inclui Kafka (lê dos brokers reais) e Auto Loader (cloudFiles, que lê do caminho real de armazenamento em nuvem). Nenhum dos dois é redirecionado para os seus dados simulados.

• Não utilize a função com valor de tabela event_log() em um teste de unidade de pipeline. No modo de teste, event_log() não é redirecionado para o log de eventos da sua execução de teste. Ele pode retornar o log de eventos de produção ou um log registrado anteriormente, de modo que as asserções realizadas sobre ele possam ler dados de produção. Em vez disso, use o event_log_table_name retornado pela execução e consulte-o por meio de test_spark. event_log_table_name pode ser None (por exemplo, se o nome da tabela de log de eventos não puder ser resolvido), portanto, verifique-o antes de consultar:

  status = test_pipeline.run(test_spark, set(["catalog.schema.table"]))
  assert status.event_log_table_name is not None
  events = test_spark.table(status.event_log_table_name)
  

  Não afirme status.is_success antes de ler o log de eventos se sua meta é diagnosticar uma atualização com falha. O log de eventos geralmente é o que você inspeciona para entender por que uma atualização falhou.

Operações de governança e DDL

• Alterações em catálogo, esquema, permissões, propriedade, tags e políticas não são compatíveis. Isso inclui CREATE/DROP/ALTER CATALOG, CREATE/DROP/ALTER SCHEMA(incluindo SET MANAGED LOCATION), GRANT/REVOKE, ALTER ... OWNER TO, SET/UNSET TAGSe .CREATE/DROP POLICY Alguns formulários SQL executados por meio test_spark são rejeitados como defesa em profundidade; outras formas ou as mesmas operações invocadas por meio de APIs diretas podem alcançar objetos de produção reais. Não confie nesses guardas como um limite de isolamento. Mantenha essas instruções fora do código de teste e fora de qualquer código de pipeline executado pelas saídas selecionadas.

Limitações operacionais

• A execução simultânea não tem suporte: não há suporte para a execução de um teste e uma atualização de pipeline ao mesmo tempo e o sistema não o impede. Não há nenhuma coordenação entre os dois, portanto, executá-los simultaneamente pode lutar por recursos, degradando severamente o desempenho da atualização de produção ou fazendo com que o teste não seja iniciado. Não inicie um teste enquanto o pipeline estiver executando uma atualização (ou inicie uma atualização enquanto um teste estiver em execução); aguarde até que qualquer atualização em andamento seja concluída antes de executar testes.
• Esquemas temporários após o encerramento anormal: cada execução de teste cria um esquema temporário (nomeado redirecting_<id>) no catálogo padrão do pipeline e o descarta automaticamente quando a execução é concluída. Se uma execução terminar de forma anormal (por exemplo, a computação será perdida no meio da execução), o esquema temporário poderá ser deixado para trás, mantendo as tabelas simuladas e de saída da execução. Isso não afeta os dados de produção. Para liberar espaço de armazenamento, exclua manualmente todos os esquemas restantes cujos nomes comecem com redirecting_ no catálogo padrão do pipeline.
• As execuções de teste consomem computação: execuções de teste são executadas na computação do pipeline e são cobradas como atualizações normais de pipeline. Não há medição separada para as execuções de teste.
• Não há suporte para a atualização completa: apenas a atualização seletiva está disponível. test_pipeline.run() atualiza as saídas que você selecionar (ou todas as saídas, caso nenhuma seleção seja feita); a atualização completa e a seleção para atualização completa não estão implementadas.

Limitações de autoria e fidelidade

• Execução somente do editor: os testes devem ser executados no Editor do Lakeflow Pipelines baseado na Web.
• Apenas testes em Python: Os testes devem ser escritos em Python. Você pode testar pipelines SQL, mas os próprios testes devem ser escritos em Python.
• Fidelidade de governança: dados simulados não herdam filtros de linha nem máscaras de coluna definidos nas tabelas de produção que substituem. Os resultados do teste refletem as entradas simuladas exatamente como você as fornece e podem diferir de como a mesma consulta se comporta em dados de produção controlados.

Etapa 1: atualizar as configurações do pipeline

Configure o pipeline para ser executado no canal PREVIEW em modo acionado.

1. Na interface do usuário, abra seu pipeline e clique em Configurações>Configurações avançadas>Canal>Visualização
2. Defina o modo do pipeline como Acionado (não use o Contínuo).

Como alternativa, edite as configurações de pipeline JSON diretamente:

"continuous": false,
"channel": "PREVIEW"

Etapa 2: Criar um arquivo de teste

No Editor do Lakeflow Pipelines, clique no + botão (adicionar) e selecione Testar. Isso cria um arquivo de teste (e a tests pasta, se ainda não existir) que não está incluído no código-fonte do pipeline. Você não precisa criar a tests pasta por conta própria.

Etapa 3: Gerar testes

O Genie Code pode gerar estrutura básica de testes:

• Dentro do arquivo de teste, clique no botão Gerar testes .

  

• Como alternativa, use /tests dentro do modo de agente do Genie Code.

  

Use o Genie Code para gerar código padrão e, em seguida, personalize-o para seus casos extremos.

Como alternativa, você pode escrever o código de teste por conta própria. Adicione as seguintes importações à parte superior de cada arquivo de teste:

import pytest
from pyspark.pipelines.testing import TestPipeline, test_spark

test_pipeline = TestPipeline.active()

Etapa 4: Executar testes

Executar testes no Editor do Lakeflow Pipelines:

• Clique no botão (reproduzir) na margem, ao lado de uma função de teste, para executar um teste individual.
• Clique em Executar testes no arquivo na parte superior do arquivo de teste para executar todos os testes nesse arquivo.

Os resultados do teste (êxito ou falha) aparecem no painel inferior do Editor. Examine os erros de declaração para depurar falhas.

Testar APIs

Criar dados fictícios

Você pode simular dados de entrada usando SQL ou createDataFrame:

# Option 1: Using SQL
test_spark.sql("""
    CREATE TABLE catalog.schema.table_name AS
    SELECT * FROM VALUES
        (1, 'value1'),
        (2, 'value2')
    AS t(id, name)
""")

# Option 2: Using createDataFrame
df = test_spark.createDataFrame(
    [(1, 'value1'), (2, 'value2')],
    schema=["id", "name"]
)
df.write.saveAsTable("catalog.schema.table_name")

Para gerar volumes maiores de dados sintéticos realistas, você pode usar a biblioteca Faker . Execute %pip install faker no seu pipeline primeiro e, em seguida, crie um DataFrame a partir de UDFs baseadas no Faker:

# Option 3: Using Faker for synthetic data
from pyspark.sql import functions as F
from faker import Faker

fake = Faker()
fake_firstname = F.udf(fake.first_name)
fake_lastname = F.udf(fake.last_name)
fake_email = F.udf(fake.ascii_company_email)

df = (
    test_spark.range(0, 100)
    .withColumn("firstname", fake_firstname())
    .withColumn("lastname", fake_lastname())
    .withColumn("email", fake_email())
)
df.write.saveAsTable("catalog.schema.table_name")

Executar o pipeline ou tabelas específicas

# Run specific tables
test_pipeline.run(test_spark, set(["catalog.schema.table1", "catalog.schema.table2"]))

# Run all tables in the pipeline
test_pipeline.run(test_spark)

Exemplos

Exemplo 1: Testando agregações com contagem de linhas, esquema e tratamento nulo

Objetivo: validar se a agregação de usuários conta corretamente os usuários por tipo, lida com e-mails nulos e gera o esquema esperado.

Transformações de pipeline:

Essas transformações criam um pipeline simples de duas tabelas: users seleciona dados do usuário e counts agrupa usuários por tipo e conta o total de usuários e emails válidos.

from pyspark import pipelines as dp
from pyspark.sql.functions import col, count, count_if

@dp.table
def users():
    return (
        spark.read.table("catalog.schema.wanderbricks_users")
        .select("user_id", "email", "name", "user_type")
    )

@dp.table
def counts():
    return (
        spark.read.table("catalog.schema.users")
        .withColumn("valid_email", col("email").isNotNull())
        .groupBy("user_type")
        .agg(
            count("user_id").alias("total_count"),
            count_if("valid_email").alias("count_valid_emails")
        )
    )

Testes:

Esses testes validam contagens de linhas, estrutura de esquema, tratamento nulo e lógica de agregação criando dados de usuário fictícios com nulos intencionais e executando o pipeline isoladamente.

import pytest
from pyspark.pipelines.testing import TestPipeline, test_spark
from pyspark.testing import assertDataFrameEqual

test_pipeline = TestPipeline.active()

# Mock data fixture
def mock_users(session):
    session.sql("""
        CREATE TABLE catalog.schema.wanderbricks_users AS
        SELECT * FROM VALUES
            (1, 'alice@example.com', 'Alice', 'admin'),
            (2, NULL, 'Bob', 'user'),
            (3, 'charlie@example.com', 'Charlie', 'user'),
            (4, NULL, 'Dana', 'admin')
        AS t(user_id, email, name, user_type)
    """)

# Test 1: Row count
def test_users_row_count(test_spark):
    mock_users(test_spark)
    test_pipeline.run(test_spark, set(["catalog.schema.users"]))
    result = test_spark.table("catalog.schema.users")
    assert result.count() == 4

# Test 2: Schema validation
def test_users_schema(test_spark):
    mock_users(test_spark)
    test_pipeline.run(test_spark, set(["catalog.schema.users"]))
    result = test_spark.table("catalog.schema.users")
    expected_fields = {"user_id", "email", "name", "user_type"}
    actual_fields = set(f.name for f in result.schema.fields)
    assert expected_fields == actual_fields

# Test 3: Null handling
def test_users_null_handling(test_spark):
    mock_users(test_spark)
    test_pipeline.run(test_spark, set(["catalog.schema.users"]))
    result = test_spark.table("catalog.schema.users")
    null_emails = result.filter("email IS NULL").count()
    assert null_emails == 2

# Test 4: Aggregation
def test_counts(test_spark):
    mock_users(test_spark)
    # Run both tables since counts depends on users
    test_pipeline.run(test_spark, set(["catalog.schema.users", "catalog.schema.counts"]))
    result = test_spark.table("catalog.schema.counts")
    # Check counts for each user_type
    admin_row = result.filter("user_type = 'admin'").collect()[0]
    user_row = result.filter("user_type = 'user'").collect()[0]
    assert admin_row["total_count"] == 2
    assert admin_row["count_valid_emails"] == 1
    assert user_row["total_count"] == 2
    assert user_row["count_valid_emails"] == 1

# Test 5: Full DataFrame comparison with assertDataFrameEqual
def test_counts_full_dataframe(test_spark):
    mock_users(test_spark)
    test_pipeline.run(test_spark, set(["catalog.schema.users", "catalog.schema.counts"]))
    result = test_spark.table("catalog.schema.counts")
    expected = test_spark.createDataFrame(
        [("admin", 2, 1), ("user", 2, 1)],
        schema=["user_type", "total_count", "count_valid_emails"]
    )
    assertDataFrameEqual(result, expected)

Exemplo 2: Testando o CDC Automático

Objetivo: Validar se o Auto CDC processa corretamente o feed de alterações com inserções e atualizações.

Transformação de pipeline:

Essa transformação configura o Auto CDC a partir de um change feed, que lê alterações em fluxo e as aplica à tabela de destino como SCD Tipo 1 (mantendo apenas a versão mais recente).

from pyspark import pipelines as dp
from pyspark.sql.functions import col

@dp.view
def users():
    return spark.readStream.table("catalog.schema.change_feed")

dp.create_streaming_table("target_autocdc")
dp.create_auto_cdc_flow(
    target="target_autocdc",
    source="users",
    keys=["userId"],
    sequence_by=col("ts"),
    stored_as_scd_type=1
)

Testes:

O primeiro teste cria um feed de alterações fictício com vários registros para o mesmo userId (simulando uma atualização) e verifica se apenas o registro mais recente é mantido no destino. O segundo teste simula eventos que chegam com atraso e fora de ordem ao executar o pipeline, anexar mais eventos ao change feed e executar o pipeline novamente.

import pytest
from pyspark.pipelines.testing import TestPipeline, test_spark

test_pipeline = TestPipeline.active()

# Test 1: Standard inserts and updates
def test_auto_cdc_flow(test_spark):
    # Create a mock change feed table
    test_spark.sql("""
        CREATE TABLE catalog.schema.change_feed AS
        SELECT * FROM VALUES
            (1, 'Alice', 1000),
            (2, 'Bob', 1001),
            (1, 'Alice Updated', 1002)
        AS t(userId, name, ts)
    """)
    # Run the pipeline
    test_pipeline.run(test_spark, set(["catalog.schema.target_autocdc"]))
    # Read the output
    result = test_spark.table("catalog.schema.target_autocdc")
    # Verify two users exist
    user_ids = set(row["userId"] for row in result.collect())
    assert user_ids == {1, 2}
    # Verify latest record for userId=1 has ts=1002
    latest_user1 = result.filter("userId = 1").collect()[0]
    assert latest_user1["ts"] == 1002
    assert latest_user1["name"] == "Alice Updated"
    # Verify userId=2 has ts=1001
    user2 = result.filter("userId = 2").collect()[0]
    assert user2["ts"] == 1001

# Test 2: Late-arriving and out-of-order events
def test_auto_cdc_late_arriving(test_spark):
    # First batch of change events
    test_spark.sql("""
        CREATE TABLE catalog.schema.change_feed AS
        SELECT * FROM VALUES
            (1, 'Alice', 1000),
            (2, 'Bob', 1001)
        AS t(userId, name, ts)
    """)
    # Run the pipeline with the initial batch
    test_pipeline.run(test_spark, set(["catalog.schema.target_autocdc"]))

    # Append late-arriving events to the change feed:
    # - A newer event for userId=1 (ts=1003) that arrived after the first run
    # - A stale event for userId=2 (ts=999) with a timestamp older than what is already applied
    test_spark.sql("""
        INSERT INTO catalog.schema.change_feed VALUES
            (1, 'Alice Updated', 1003),
            (2, 'Bob (stale)', 999)
    """)
    # Re-run the pipeline. sequence_by=ts ensures stale events do not overwrite newer state.
    test_pipeline.run(test_spark, set(["catalog.schema.target_autocdc"]))

    result = test_spark.table("catalog.schema.target_autocdc")
    # userId=1 should reflect the newer late-arriving event
    alice = result.filter("userId = 1").collect()[0]
    assert alice["ts"] == 1003
    assert alice["name"] == "Alice Updated"
    # userId=2 should be unchanged: the stale event with an older ts is ignored
    bob = result.filter("userId = 2").collect()[0]
    assert bob["ts"] == 1001
    assert bob["name"] == "Bob"

Exemplo 3: Testando o CDC automático a partir de snapshot

Objetivo: Valide se o CDC processa corretamente as alterações no snapshot, incluindo inserções, atualizações e exclusões.

Transformação do pipeline:

Essa transformação configura o Auto CDC a partir de um snapshot, lendo de uma tabela de snapshot e rastreando alterações ao longo do tempo como SCD Tipo 2 (mantendo o histórico completo).

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
    return spark.read.table("catalog.schema.snapshot")

dp.create_streaming_table("catalog.schema.target")
dp.create_auto_cdc_from_snapshot_flow(
    target="target",
    source="source",
    keys=["userId"],
    stored_as_scd_type=2
)

Teste:

Esse teste cria um instantâneo inicial, executa o pipeline e simula uma atualização do instantâneo truncando e inserindo novos dados, para verificar se o CDC captura todas as alterações.

import pytest
from pyspark.pipelines.testing import TestPipeline, test_spark

test_pipeline = TestPipeline.active()

def test_auto_cdc_from_snapshot_flow(test_spark):
    # Create initial snapshot
    test_spark.sql("""
        CREATE TABLE catalog.schema.snapshot AS
        SELECT * FROM VALUES
            (1, 'Alice', '2024-01-01'),
            (2, 'Bob', '2024-01-02')
        AS t(userId, name, created_at)
    """)
    # Run the pipeline
    test_pipeline.run(test_spark, set(["catalog.schema.target"]))
    # Simulate a new snapshot by truncating and inserting updated data
    test_spark.sql("TRUNCATE TABLE catalog.schema.snapshot")
    test_spark.sql("INSERT INTO catalog.schema.snapshot VALUES (2, 'Bob', '2024-01-03')")
    test_pipeline.run(test_spark, set(["catalog.schema.target"]))
    # Verify SCD Type 2: should have 3 rows (original Alice, original Bob, updated Bob)
    result = test_spark.table("catalog.schema.target")
    assert result.count() == 3
    user_ids = [row["userId"] for row in result.collect()]
    assert set(user_ids) == {1, 2}

Exemplo 4: testando junções e expectativas

Objetivo: Validar se as junções funcionam corretamente e se as expectativas filtram dados inválidos.

Transformação do pipeline:

Essa transformação combina imagens da propriedade com comodidades e aplica um critério para filtrar imagens enviadas antes de janeiro de 2024.

from pyspark import pipelines as dp

@dp.table
@dp.expect_or_drop("uploaded after Jan 2024", "uploaded_at > '2024-01-01'")
def property_images_amenities_join():
    return (
        spark.read.table("catalog.schema.property_images")
        .join(
            spark.read.table("catalog.schema.property_amenities"),
            on="property_id",
            how="inner"
        )
    )

Testes:

Esses testes verificam se a junção produz o número correto de linhas e se a expectativa filtra com êxito os registros com datas de carregamento inválidas.

import pytest
from pyspark.pipelines.testing import TestPipeline, test_spark

test_pipeline = TestPipeline.active()

# Mock property datasets
def mock_properties(session):
    session.sql("""
        CREATE TABLE catalog.schema.property_images AS
        SELECT * FROM VALUES
            (101, 'img1.jpg', '2024-02-01'),
            (102, 'img2.jpg', '2024-01-15'),
            (103, 'img3.jpg', '2024-12-20')
        AS t(property_id, image_url, uploaded_at)
    """)
    session.sql("""
        CREATE TABLE catalog.schema.property_amenities AS
        SELECT * FROM VALUES
            (101, 'wifi'),
            (102, 'pool'),
            (103, 'parking')
        AS t(property_id, amenity)
    """)

# Test 1: Join
def test_property_join(test_spark):
    mock_properties(test_spark)
    test_pipeline.run(test_spark, set(["catalog.schema.property_images_amenities_join"]))
    result = test_spark.table("catalog.schema.property_images_amenities_join")
    # Should have 3 rows after join
    assert result.count() == 3
    # Check all property_ids are present
    property_ids = set(row["property_id"] for row in result.collect())
    assert property_ids == {101, 102, 103}

# Test 2: Expectation
def test_property_expectation(test_spark):
    mock_properties(test_spark)
    # Add a row with uploaded_at before Jan 2024
    test_spark.sql("""
        INSERT INTO catalog.schema.property_images VALUES (104, 'img4.jpg', '2023-12-31')
    """)
    # Add a matching row in the amenities table for the join
    test_spark.sql("""
        INSERT INTO catalog.schema.property_amenities VALUES (104, 'gym')
    """)
    test_pipeline.run(test_spark, set(["catalog.schema.property_images_amenities_join"]))
    result = test_spark.table("catalog.schema.property_images_amenities_join")
    # Only property_ids with uploaded_at > '2024-01-01' should be present
    valid_ids = set(row["property_id"] for row in result.collect())
    assert 104 not in valid_ids
    assert valid_ids == {101, 102, 103}