Transformar dados com o Delta Live Tables

Este artigo descreve como você pode utilizar o Delta Live Tables para declarar transformações em conjuntos de dados e especificar como os registros são processados através da logica de consulta. Ele também contém alguns exemplos de padrões de transformação comuns que podem ser úteis ao criar os pipelines do Delta Live Tables.

Você pode definir um conjunto de dados em relação a qualquer consulta que retorne um DataFrame. Você pode utilizar operações internas do Apache Spark, UDFs, logica personalizada e modelos do MLflow como transformações no seu pipeline do Delta Live Tables. Uma vez que os dados tenham sido ingeridos no seu pipeline do Delta Live Tables, você poderá definir novos conjuntos de dados em relação às fontes upstream para criar novas tabelas de streaming, visualizações materializadas e exibições.

Para saber como executar com eficácia o processamento com estado com Delta Live Tables, confira Otimizar o processamento com estado em Delta Live Tables com marcas d'água.

Quando usar exibições, exibições materializadas e tabelas de streaming

Para garantir que seus pipelines sejam eficientes e passíveis de manutenção, escolha o melhor tipo de conjunto de dados ao implementar suas consultas de pipeline.

Considere o uso de uma exibição quando:

• Você tiver uma consulta grande ou complexa que deseja dividir em consultas mais fáceis de gerenciar.
• Você quiser validar resultados intermediários usando expectativas.
• Você quiser reduzir os custos de armazenamento e computação e não precisar da materialização dos resultados da consulta. Como as tabelas são materializadas, elas requerem recursos adicionais de computação e armazenamento.

Considere o uso de uma exibição materializada quando:

• Diversas consultas downstream consumirem a tabela. Como as exibições são calculadas sob demanda, a exibição é recalculada toda vez que é consultada.
• Outros pipelines, trabalhos ou consultas consomem a tabela. Como as exibições não são materializadas, só é possível usá-las no mesmo pipeline.
• Você quiser exibir os resultados de uma consulta durante o desenvolvimento. Como as tabelas são materializadas e podem ser exibidas e consultadas fora do pipeline, o uso de tabelas durante o desenvolvimento pode ajudar a validar a exatidão dos cálculos. Após a validação, converta as consultas que não requerem materialização em exibições.

Considere o uso de uma tabela de streaming quando:

• Uma consulta é definida em relação a uma fonte de dados que está crescendo de forma contínua ou incremental.
• Os resultados da consulta devem ser computados de forma incremental.
• Deseja-se alta taxa de transferência e baixa latência para o pipeline.

Observação

As tabelas de streaming são sempre definidas em relação a fontes de streaming. Você também pode utilizar fontes de streaming com APPLY CHANGES INTO para aplicar atualizações de feeds da CDA. Confira API APPLY CHANGES: Simplifique a captura de dados de alterações nas Tabelas Dinâmicas Delta.

Combinar tabelas de streaming e exibições materializadas em um único pipeline

As tabelas de streaming herdam as garantias de processamento do Streaming Estruturado do Apache Spark e são configuradas para processar consultas de fontes de dados somente acréscimo, em que novas linhas são sempre inseridas na tabela de origem, ao invés de modificadas.

Observação

Embora, por padrão, as tabelas de streaming exijam fontes de dados somente acréscimo, quando uma fonte de streaming é outra tabela de streaming que requer atualizações ou exclusões, é possível substituir esse comportamento pelo sinalizador skipChangeCommits.

Um padrão comum de streaming inclui a ingestão de dados de origem para criar os conjuntos de dados iniciais em um pipeline. Esses conjuntos de dados iniciais são comumente chamados de tabelas bronze e geralmente executam transformações simples.

Por outro lado, as tabelas finais em um pipeline, comumente chamadas de tabelas ouro, geralmente exigem agregações complicadas ou leitura de fontes que são o destino de uma operação APPLY CHANGES INTO. Como essas operações criam inerentemente atualizações em vez de acrescentar, elas não têm suporte como entradas para tabelas de streaming. Essas transformações são mais adequadas para exibições materializadas.

Ao combinar as tabelas de streaming e as exibições materializadas em um único pipeline, você pode simplificar seu pipeline, evitar a dispendiosa reingestão ou o reprocessamento de dados brutos e ter todo o poder do SQL para computar agregações complexas em um conjunto de dados eficientemente codificado e filtrado. O exemplo a seguir ilustra esse tipo de processamento misto:

Observação

Esses exemplos utilizam o Carregador Automático para carregar arquivos do armazenamento em nuvem. Para carregar arquivos com o Carregador Automático em um pipeline habilitado para o Catálogo do Unity, você deve usar localizações externas. Para saber mais sobre como usar o Catálogo do Unity com o Delta Live Tables, confira Usar o Catálogo do Unity com seus pipelines do Delta Live Tables.

Python

@dlt.table
def streaming_bronze():
  return (
    # Since this is a streaming source, this table is incremental.
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load("abfss://path/to/raw/data")
  )

@dlt.table
def streaming_silver():
  # Since we read the bronze table as a stream, this silver table is also
  # updated incrementally.
  return dlt.read_stream("streaming_bronze").where(...)

@dlt.table
def live_gold():
  # This table will be recomputed completely by reading the whole silver table
  # when it is updated.
  return dlt.read("streaming_silver").groupBy("user_id").count()

SQL

CREATE OR REFRESH STREAMING TABLE streaming_bronze
AS SELECT * FROM cloud_files(
  "abfss://path/to/raw/data", "json"
)

CREATE OR REFRESH STREAMING TABLE streaming_silver
AS SELECT * FROM STREAM(LIVE.streaming_bronze) WHERE...

CREATE OR REFRESH LIVE TABLE live_gold
AS SELECT count(*) FROM LIVE.streaming_silver GROUP BY user_id

Saiba mais sobre como usar o Carregador Automático para ler com eficiência arquivos JSON do armazenamento do Azure para processamento incremental.

Junções estáticas de fluxo

As junções estáticas de fluxo são uma boa escolha ao desnormalizar um fluxo contínuo de dados somente acréscimo com uma tabela de dimensão principalmente estática.

A cada atualização do pipeline, os novos registros do fluxo são ingressados com o instantâneo mais atual da tabela estática. Se os registros forem adicionados ou atualizados na tabela estática depois que os dados correspondentes da tabela de streaming tiverem sido processados, os registros resultantes não serão recalculados, a menos que uma atualização completa seja executada.

Nos pipelines configurados para execução disparada, a tabela estática retorna resultados a partir do tempo em que a atualização foi iniciada. Nos pipelines configurados para execução contínua, toda vez que a tabela processa uma atualização, a versão mais recente da tabela estática é consultada.

A seguir, um exemplo de uma junção estática de fluxo:

Python

@dlt.table
def customer_sales():
  return dlt.read_stream("sales").join(dlt.read("customers"), ["customer_id"], "left")

SQL

CREATE OR REFRESH STREAMING TABLE customer_sales
AS SELECT * FROM STREAM(LIVE.sales)
  INNER JOIN LEFT LIVE.customers USING (customer_id)

Calcular eficientemente as agregações

Você pode utilizar tabelas de streaming para calcular de forma incremental agregações distributivas simples, como contagem, mínimo, máximo ou soma, e agregações algébricas, como média ou desvio padrão. A Databricks recomenda a agregação incremental para consultas com um número limitado de grupos, por exemplo, uma consulta com cláusula GROUP BY country. Somente novos dados de entrada são lidos a cada atualização.

Para saber mais sobre como escrever consultas do Delta Live Tables que executam agregações incrementais, confira Executar agregações em janelas com marcas d'água.

Usar modelos de MLFlow em um pipeline do Delta Lives Table

Observação

Para usar modelos do MLflow em um pipeline habilitado para Catálogo do Unity, o pipeline precisa ser configurado para usar o canal preview. Para usar o canal current, você precisa configurar o pipeline para publicar no metastore do Hive.

Você pode utilizar modelos treinados pelo MLflow nos pipelines do Delta Live Tables. Os modelos MLflow são tratados como transformações no Azure Databricks, o que significa que eles agem sobre uma entrada do DataFrame do Spark e retornam resultados como um DataFrame do Spark. Como o Delta Live Tables define os conjuntos de dados em relação aos DataFrames, você pode converter as cargas de trabalho do Apache Spark que utilizam o MLflow para o Delta Live Tables com apenas algumas linhas de código. Para saber mais sobre o MLflow, confira Gerenciamento do ciclo de vida do ML usando o MLflow.

Se você já tiver um notebook Python chamando um modelo MLflow, poderá adaptar esse código ao Delta Live Tables utilizando o decorador @dlt.table e garantindo que as funções sejam definidas para retornar os resultados da transformação. O Delta Live Tables não instala o MLflow por padrão, portanto, verifique se você %pip install mlflow e importe mlflow e dlt na parte superior do notebook. Para obter uma introdução à sintaxe Delta Live Tables, consulte Exemplo: Ingerir e processar dados de nomes de bebê de Nova York.

Para utilizar os modelos do MLflow nas tabelas do Delta Live, conclua as etapas a seguir:

1. Obtenha a ID de execução e o nome do modelo do modelo de MLflow. A ID de execução e o nome do modelo são usados para construir o URI do modelo de MLflow.
2. Use o URI para definir uma UDF do Spark para carregar o modelo de MLflow.
3. Chame a UDF em suas definições de tabela para usar o modelo de MLflow.

O exemplo a seguir mostra a sintaxe básica para esse padrão:

%pip install mlflow

import dlt
import mlflow

run_id= "<mlflow-run-id>"
model_name = "<the-model-name-in-run>"
model_uri = f"runs:/{run_id}/{model_name}"
loaded_model_udf = mlflow.pyfunc.spark_udf(spark, model_uri=model_uri)

@dlt.table
def model_predictions():
  return dlt.read(<input-data>)
    .withColumn("prediction", loaded_model_udf(<model-features>))

Como um exemplo concluído, o código a seguir define um Spark UDF chamado loaded_model_udf que carrega um modelo MLflow treinado em dados de risco de empréstimo. As colunas de dados utilizadas para fazer a previsão são passadas como um argumento para o UDF. A tabela loan_risk_predictions calcula as previsões para cada linha em loan_risk_input_data.

%pip install mlflow

import dlt
import mlflow
from pyspark.sql.functions import struct

run_id = "mlflow_run_id"
model_name = "the_model_name_in_run"
model_uri = f"runs:/{run_id}/{model_name}"
loaded_model_udf = mlflow.pyfunc.spark_udf(spark, model_uri=model_uri)

categoricals = ["term", "home_ownership", "purpose",
  "addr_state","verification_status","application_type"]

numerics = ["loan_amnt", "emp_length", "annual_inc", "dti", "delinq_2yrs",
  "revol_util", "total_acc", "credit_length_in_years"]

features = categoricals + numerics

@dlt.table(
  comment="GBT ML predictions of loan risk",
  table_properties={
    "quality": "gold"
  }
)
def loan_risk_predictions():
  return dlt.read("loan_risk_input_data")
    .withColumn('predictions', loaded_model_udf(struct(features)))

Manter as exclusões ou atualizações manuais

O Delta Live Tables permite excluir ou atualizar manualmente os registros de uma tabela e fazer uma operação de atualização para recomputar as tabelas de downstream.

Por padrão, o Delta Live Tables recomputa os resultados da tabela com base nos dados de entrada sempre que um pipeline é atualizado, portanto, certifique-se de que o registro excluído não seja recarregado a partir dos dados de origem. A definição da propriedade da tabela pipelines.reset.allowed como false impede a atualização de uma tabela, mas não impede gravações incrementais nas tabelas nem impede o fluxo de novos dados para a tabela.

O diagrama a seguir ilustra um exemplo usando duas tabelas de streaming:

• Ingerir raw_user_table dados brutos do usuário de uma fonte.
• bmi_table computa incrementalmente pontuações de IMC usando peso e altura de raw_user_table.

Você deseja excluir ou atualizar manualmente os registros de usuários do raw_user_table e recomputar o bmi_table.

O código a seguir demonstra a configuração da propriedade da tabela pipelines.reset.allowed como false para desabilitar a atualização completa para raw_user_table, de modo que as alterações pretendidas sejam mantidas ao longo do tempo, mas as tabelas de downstream sejam recomputadas quando uma atualização do pipeline for executada:

CREATE OR REFRESH STREAMING TABLE raw_user_table
TBLPROPERTIES(pipelines.reset.allowed = false)
AS SELECT * FROM cloud_files("/databricks-datasets/iot-stream/data-user", "csv");

CREATE OR REFRESH STREAMING TABLE bmi_table
AS SELECT userid, (weight/2.2) / pow(height*0.0254,2) AS bmi FROM STREAM(LIVE.raw_user_table);