Noções básicas do Spark

Conceitos fundamentais subjacentes ao dimensionamento, otimização e resolução de problemas. Leia isso primeiro se você é novo no Spark in Fabric.

Prós e contras gerais

Cenário: você é novo no Spark. O que são os Prós e os Contras

Caso de uso	Melhores práticas
Usar formatos serializados otimizados	Prefira formatos como Avro, Parquet ou ORC (Optimized Row Columnar) uma vez que incorporam esquemas, são compactos e otimizam o armazenamento e o processamento. No Fabric, use o formato Delta para garantias de Atomicidade, Consistência, Isolamento, Durabilidade (ACID) e vantagens de desempenho.
Seja cauteloso com XML/JSON	Não confie na inferência de esquema para grandes arquivos JSON (JavaScript Object Notation) ou XML (Extensible Markup Language), pois o Spark lê todo o conjunto de dados para inferir o esquema, o que retarda o processamento e consome memória intensamente. Forneça um esquema primário estático ao ler JSON/XML ou use `.option("samplingRatio", 0.1)` para acelerar leituras, mas esteja ciente de que, se o exemplo não representar o conjunto de dados completo, as leituras poderão falhar. Uma abordagem mais segura infere o esquema de uma amostra representativa e persiste-o para todas as leituras. Evite analisar arquivos XML grandes. A análise XML é inerentemente mais lenta devido ao processamento de tags e à conversão de tipos.
Otimizar junções e filtragem	Faça: Aplique a poda de colunas e a filtragem ao nível de linha antes das junturas para reduzir o uso de shuffle e de memória. O otimizador Catalyst lida automaticamente com o pushdown de predicados quando o utilizador utiliza APIs de DataFrame. Evite APIs RDD (Resilient Distributed Dataset) porque elas ignoram as otimizações do Catalyst.
Prefira DataFrames em vez de RDDs	Utilize DataFrames em vez de RDDs para a maioria das operações. Os DataFrames usam o otimizador Catalyst e o mecanismo de execução Tungsten para uma execução eficiente.
Habilitar a execução adaptável de consultas (AQE)	Fazer: ative o AQE para otimizar dinamicamente partições aleatórias e lidar com dados distorcidos automaticamente.

Gerenciamento de memória do executor

Cenário: Você deseja entender o gerenciamento de memória do executor para ajuste de desempenho.

Mesmo que um executor esteja configurado com 56 GB de memória, o Spark não permite que todo ele seja usado diretamente para dados do usuário. O Spark Core divide e gerencia a memória do executor:

Memória reservada: Uma parte fixa reservada para a sobrecarga interna do sistema e do Spark (por exemplo, Java Virtual Machine (JVM), componentes internos).
Memória do usuário: Armazena funções definidas pelo usuário (UDFs), variáveis locais, estruturas de dados (listas, mapas, dicionários) e objetos criados durante a computação.
Memória de armazenamento: Contém dados em cache/persistidos, variáveis de difusão e dados de intercâmbio que podem ser armazenados em cache.
Memória de execução: Usado para computação intermediária (shuffles, junções, ordenações, agregações).
Compartilhamento dinâmico de memória: O limite entre a memória de armazenamento e a memória de execução é móvel. O Spark pode emprestar memória de uma região para outra, permitindo o uso flexível da memória.
Derrame: Ocorre quando a demanda de memória de Armazenamento ou Execução excede a memória disponível após o empréstimo. Isso força os dados para o disco, o que pode afetar o desempenho.

Erros de Falta de Memória (OOM)

Cenário: As tarefas do Spark falham com erros de falta de memória (Out of Memory, OOM).

Driver OOM:

Erros de OOM do driver ocorrem quando o driver do Spark excede a sua memória alocada.

Causa comum: operações pesadas do driver, como collect(), countByKey(), ou chamadas grandes toPandas() que transferem muitos dados para a memória do driver.

Mitigação: Evite operações pesadas com o motorista sempre que possível. Se for inevitável, aumente o tamanho do driver e o benchmark para encontrar a configuração ideal.

Executor Fora de Memória (OOM):

Os erros OOM do executor ocorrem quando um executor do Spark excede a memória que lhe foi alocada.

Causa comum: Transformações que utilizam intensivamente a memória e a computação em grandes conjuntos de dados (por exemplo, junções amplas, agregações, embaralhamentos) ou conjuntos de dados armazenados em cache ou persistentes que excedem a memória disponível do executor (execução + regiões de armazenamento).

Atenuação: aumente a memória do executor, se necessário, ajuste as frações de memória do Spark (spark.memory.fraction, spark.memory.storageFraction) e persista seletivamente. Certifique-se de que os dados armazenados em cache se encaixam na memória disponível.

Desvios de Distribuição de Dados

Sintomas de distorção:

Algumas tarefas levam mais tempo do que outras na interface do usuário do Spark (as tarefas de palco mostram cauda pesada).
Grande diferença entre os tempos medianos e máximos das tarefas nas métricas de estágio.
Estágios com grandes volumes de leitura ou escrita de shuffle para algumas partições.

Causas comuns:

Distribuição de dados irregular para as chaves de junção/grupo (chaves quentes).
Particionamento incorreto ou poucas partições para o volume de dados.
Anomalias de dados upstream que produzem registros grandes ou muitas chaves nulas/vazias.

Atenuação:

Reparticione ou coalesca para aumentar o paralelismo de partição e equilibrar os tamanhos.
Aplique 'salting' de chaves ou particionamento personalizado para distribuir chaves quentes entre partições.
Use AQE (Adaptive Query Execution) para unir partições após o shuffle e habilitar otimizações de junção com desvio.
Utilize uniões de difusão para pequenas tabelas de pesquisa para evitar completamente redistribuições.
Persista conjuntos de dados intermediários balanceados antes de estágios caros e execute novamente o trabalho.

Práticas recomendadas da UDF

Cenário: Você precisa aplicar lógica personalizada que não pode ser expressa por meio de funções internas do DataFrame.

Use APIs do Spark DataFrame sempre que possível. O otimizador Catalyst otimiza funções internas e as executa nativamente na JVM, para que ofereçam o melhor desempenho.

Se for preciso usar uma UDF (Função Definida pelo Usuário), evite utilizar as UDF regulares do PySpark em Python. Em vez disso, considere as seguintes alternativas:

Pandas UDFs (também conhecidas como UDFs vetorizadas): Use a seta Apache para transferência eficiente de dados entre JVM e Python. Os Pandas UDFs permitem operações vetorizadas, melhorando significativamente o desempenho em comparação com UDFs Python linha por linha.
UDFs Scala/Java: Execute diretamente na JVM, evitando a sobrecarga de serialização do Python. UDFs Scala/Java normalmente superam UDFs Python.

Seja cauteloso com UDFs Python. Cada executor inicia um processo Python separado, exigindo serialização e desserialização de dados entre a JVM e o Python. Isso cria um gargalo de desempenho, especialmente em escala.

Registo de erros

Cenário: Práticas recomendadas para registro de erros no Fabric Spark

Utilize log4j em vez de print(), que sobrecarrega o condutor significativamente. Com log4j, pode aceder aos logs nos logs do driver e pesquisá-los (usando o nome do logger, por exemplo: PySparkLogger).
Envolva leituras, gravações e transformações em blocos try and except . Use logger.error para exceções e logger.info para mensagens de progresso.
- Registro em Python: Ideal para registrar operações, atualizações de status ou depurar informações de código que é executado somente no Spark Driver. O módulo de logging do Python não é propagado para os logs do executor. Consulte a documentação de desenvolvimento, execução e gerenciamento de blocos de anotações.
- Spark log4j: O padrão para registo de aplicações robusto a nível de produção no Spark ao integrar-se nativamente aos logs do driver/executor do Spark.
Exemplo de uso do log4j no PySpark:
```
import traceback
# Get log4j logger
log4jLogger = spark._jvm.org.apache.log4j
logger = log4jLogger.LogManager.getLogger("PySparkLogger")
logger.info("Application started.")
try:
    # Create DataFrame with 20 records
    data = [(f"Name{i}", i) for i in range(1, 21)]  # 20 records
    df = spark.createDataFrame(data, ["name", "age"])
    logger.info("DataFrame created successfully with 20 records.")
    df.show(s)  # 's' is not defined -> will throw error but the application will not fail
except Exception as e:
    logger.error(f"Error while creating or showing DataFrame: {str(e)}\n{traceback.format_exc()}")
```
Centralize o monitoramento de erros:
- Use a extensão do emissor de diagnóstico (Monitorar aplicativos Apache Spark com o Azure Log Analytics) no ambiente e anexe aos Notebooks que executam aplicativos Spark. O emissor pode enviar logs de eventos, logs personalizados (como log4j) e métricas para o Azure Log Analytics/Azure Storage/Azure Event Hubs. Passe o nome log4j para a propriedade: spark.synapse.diagnostic.emitter.\<destination\>.filter.loggerName.match.
- Além disso, para depuração, você também pode coletar linhas/registros com falha em tabelas Lakehouse (LH) para captura de dados incorretos em nível de registro.

Comentários

Esta página foi útil?

Last updated on 2025-10-23