UDFs em Python, UDFs em Scala e tipos de dados complexos no motor de execução nativo

O motor de execução nativo do Microsoft Fabric agora suporta funções definidas pelo utilizador (UDFs) em Python, UDFs Scala e tipos de dados complexos (arrays, maps e structs). Estas capacidades permitem-lhe escrever aplicações expressivas para o Spark sem sacrificar o desempenho.

Suporte a UDF em Python

Python é uma das linguagens mais populares em engenharia de dados e ciência de dados. Historicamente, os UDFs Python introduziam uma sobrecarga significativa no Spark devido aos custos de serialização entre os processos de trabalho JVM e Python. O motor de execução nativo minimiza estas transições dispendiosas, permitindo uma execução mais rápida sem alterações de código.

Como funcionam os UDFs em Python num motor de execução nativo

Num modelo convencional de execução Spark, a execução Python UDF envolve:

  1. Conversão de dados a partir do formato interno do Spark.
  2. Serialização e transferência para processos de trabalho em Python.
  3. Execução de UDFs em Python.
  4. Serialização dos resultados de volta para a JVM.
  5. Spark retoma a execução.

Esta transferência entre ambientes de execução acarreta custos de serialização/desserialização, utilização ineficiente da CPU e pipelines de execução colunar interrompidos. O motor de execução nativo reduz esta sobrecarga ao otimizar o caminho de transferência de dados e manter o processamento vetorializado sempre que possível.

Tipos de UDF em Python suportados

O motor de execução nativo suporta:

  • Scalar UDFs: Funções de Python linha a linha registadas com udf().
  • UDFs vetorizadas (Pandas): Funções decoradas com @pandas_udf, que operam sobre lotes de dados utilizando o Apache Arrow para uma transferência eficiente.

Os UDFs vetorizados apresentam maiores ganhos de desempenho porque se alinham naturalmente com o modelo de processamento colunar do motor de execução nativo.

Exemplo: UDF vetorizado em Python

import pandas as pd
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType

@pandas_udf(DoubleType())
def calculate_discount(price: pd.Series, rate: pd.Series) -> pd.Series:
    return price * (1 - rate)

df = spark.table("sales.transactions")
result = df.withColumn("discounted_price", calculate_discount(df.price, df.discount_rate))
result.show()

Não é necessária qualquer configuração adicional além de ativar o motor de execução nativo. Os UDFs em Python existentes beneficiam automaticamente.

Suporte ao UDF da Scala

O motor de execução nativo também acelera os UDFs Scala. Como os UDFs Scala correm nativamente na JVM, o motor pode descarregar as operações suportadas para o caminho de execução vetorizado em C++, mantendo a avaliação do UDF Scala eficiente dentro do mesmo tempo de execução.

Exemplo: Scala UDF

import org.apache.spark.sql.functions.udf

val toUpperCase = udf((s: String) => s.toUpperCase)
val df = spark.table("catalog.customers")
val result = df.withColumn("name_upper", toUpperCase(df("name")))
result.show()

Os UDFs Scala que operam com tipos de dados suportados são acelerados sem alterações de código quando o motor de execução nativo está ativado.

Suporte a tipos de dados complexos

As arquiteturas modernas de casas de lago dependem de dados semi-estruturados e aninhados. O motor de execução nativo oferece agora suporte otimizado para:

Tipo de dados Description Exemplo de caso de uso
Matriz Coleção ordenada de elementos Etiquetas de eventos, categorias de produtos
Mapa Os pares chave-valor Propriedades de configuração, metadados
Struct Campos nomeados com diferentes tipos Registos de clientes encadeados, objetos de endereços

Operações suportadas para tipos complexos

O motor de execução nativo acelera operações comuns em tipos de dados complexos:

  • Funções do array: explode, array_contains, size, flatten, transform
  • Funções de mapeamento: map_keys, map_values, element_at
  • Acesso a struct: acesso ao campo por notação de ponto, getField
  • Combinações aninhadas: matrizes de estruturas, mapas com valores em matriz

Exemplo: Trabalhar com matrizes e estruturas

from pyspark.sql.functions import explode, col, size

# Read data with nested schema
df = spark.table("events.telemetry")

# Operations on arrays - accelerated by native engine
result = (df
    .filter(size(col("tags")) > 0)
    .select(
        col("event_id"),
        col("metadata.source"),  # Struct field access
        explode(col("tags")).alias("tag")
    )
)
result.show()

Exemplo: Trabalhar com mapas

from pyspark.sql.functions import map_keys, map_values, col

df = spark.table("config.settings")

# Map operations - accelerated by native engine
result = (df
    .select(
        col("setting_id"),
        map_keys(col("properties")).alias("keys"),
        map_values(col("properties")).alias("values")
    )
)
result.show()

Resultados de desempenho

O benchmarking interno demonstra melhorias significativas em cargas de trabalho que utilizam UDFs Python e tipos de dados complexos:

Tipo de carga de trabalho Melhoria do desempenho
UDFs vetorizados em Python Até 5,76x mais rápido
UDFs escalares de Python Até 1,08x mais rápido
TPC-DS de ponta a ponta (com tipos complexos) Até 2,35 vezes mais rápido

Estes ganhos resultam da redução da sobrecarga de serialização, melhora da vetorização e execução colunar de ponta a ponta.

Benefícios para padrões avançados de casas de lago

A aceleração de tipos de dados complexos é especialmente importante para:

  • Otimização Z-ORDER: As colunas aninhadas participam na disposição otimizada dos dados.
  • Agrupamento de líquidos: Colunas de tipo complexo beneficiam do agrupamento sem achatamento.
  • Análise semiestruturada: os payloads JSON e os fluxos de eventos mantêm-se aninhados para consulta natural.
  • Arquiteturas orientadas a eventos: Os dados de telemetria e IoT mantêm a sua estrutura hierárquica.

Em vez de achatar dados ou reestruturar pipelines para desempenho, trabalhe naturalmente com esquemas complexos mantendo uma elevada eficiência de execução.

Ativar a funcionalidade

Python UDF, Scala UDF e suporte a tipos de dados complexos está disponível quando o motor de execução nativo está ativado. Nenhuma configuração adicional é necessária.

Para ativar o motor de execução nativo, veja Motor de execução nativo para Engenharia de Dados de Fabrico.

Pré-requisitos

Limitações

  • Nem todas as bibliotecas Python são suportadas dentro do caminho vetorizado. As bibliotecas que requerem a serialização arbitrária de objetos Python podem ainda assim acionar o mecanismo de recurso.
  • Tipos de dados complexos profundamente aninhados (por exemplo, matrizes de mapas de estruturas) podem reverter para o motor JVM para determinadas operações.
  • O modo ANSI não é suportado pelo motor de execução nativo.

Conteúdo relacionado

  • Last updated on