Noções básicas do Spark

Conceitos fundamentais que sustentam o dimensionamento, a otimização e a solução de problemas. Leia isso primeiro se você for novo no Spark no Fabric.

Práticas e Evitações Gerais

Cenário: você é novo no Spark. O que são os Dos e Don'ts

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 memória de 56 GB, o Spark não permite que tudo seja usado diretamente para dados do usuário. O Spark Core divide e gerencia a memória do executor:

• Memória Reservada: Uma porção fixa reservada para sobrecarregas internas do sistema e do Spark (por exemplo, Máquina Virtual Java (JVM), internas).

• Memória do usuário: Armazena UDFs (Funções Definidas pelo Usuário), variáveis locais, estruturas de dados (listas, mapas, dicionários) e objetos criados durante a computação.

• Memória de armazenamento: Contém dados armazenados em cache/persistentes, variáveis de difusão e dados de shuffle que podem ser armazenados em cache.

• Memória de execução: Utilizada na computação intermediária (embaralhamentos, junções, classificações, agregações).

• Compartilhamento dinâmico de memória: O limite entre memória de Armazenamento e Execução é móvel. O Spark pode emprestar memória de uma região para a outra, permitindo o uso flexível de memória.

• Transbordo: Ocorre quando a demanda de memória para armazenamento ou execução excede a memória disponível após o empréstimo. Isso força os dados ao disco, o que pode afetar o desempenho.

  

Erros de OOM (memória insuficiente)

Cenário: os trabalhos do Spark falham com erros de memória insuficiente (OOM).

Driver OOM:

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

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

Mitigação: evite operações de driver pesado sempre que possível. Se for inevitável, aumente o tamanho do driver e o parâmetro de comparação para encontrar a configuração ideal.

Executor Sem Memória (OOM):

Erros de OOM do executor ocorrem quando um executor do Spark excede sua memória alocada.

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

Mitigação: aumente a memória do executor, se necessário, configure as frações de memória do Spark (spark.memory.fraction, spark.memory.storageFraction) e persista seletivamente. Verifique se os dados armazenados em cache se ajustam à memória disponível.

Distorções de dados

Sintomas de distorção:

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

Causas comuns:

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

Mitigação:

• Reparticionar ou coalescer para aumentar o paralelismo de partição e equilibrar os tamanhos.
• Aplique o sal de chaves ou o particionamento personalizado para espalhar teclas quentes entre partições.
• Use o AQE (Execução de Consulta Adaptável) para unir partições pós-embaralhadas e habilitar otimizações de junção de distorção.
• Use junções de broadcast para pequenas tabelas de pesquisa para evitar completamente os embaralhamentos.
• Persista conjuntos de dados intermediários equilibrados antes dos estágios caros e execute novamente o trabalho.

Práticas recomendadas da UDF

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

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

Se você precisar usar uma UDF (Função Definida pelo Usuário), evite UDFs regulares do PySpark Python. Em vez disso, considere as seguintes alternativas:

• UDFs do Pandas (também conhecidos como UDFs vetorizados): use o Apache Arrow para transferência eficiente de dados entre JVM e Python. As UDFs do Pandas permitem operações vetorizadas, melhorando significativamente o desempenho em comparação com UDFs do Python linha a linha.

• UDFs do Scala/Java: executados diretamente na JVM, evitando a sobrecarga de serialização do Python. As UDFs do Scala/Java normalmente superam as UDFs do Python.

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

Log de erros

Cenário: práticas recomendadas para registro de erros em log no Fabric Spark

1. Use log4j em vez do print() que sobrecarrega muito o driver. Com log4j, você pode acessar os logs do driver e pesquisá-los (usando o nome do logger, por exemplo: PySparkLogger).

   

2. Encapsule as leituras, gravações e transformações em os blocos try e except. Use logger.error para exceções e logger.info para mensagens de progresso.

   • Log do Python: Ideal para operações de log, atualizações de status ou informações de depuração do código que é executado somente no Driver Spark. O módulo de logging do Python não se propaga para os logs do executor. Consulte a documentação de desenvolvimento, execução e gerenciamento de notebooks.

   • Log4j do Spark: O padrão para registro em log de aplicativos robustos em ambiente de produção no Spark, pois se integra nativamente aos logs de 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()}")
   

3. Centralizar o monitoramento de erros:

   • Use a extensão do emitter de diagnóstico (Monitorar aplicativos Apache Spark com o Azure Log Analytics) no ambiente e anexe-se 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, Armazenamento do Azure ou Hubs de Eventos do Azure. 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 falhas nas tabelas do Lakehouse (LH) para captura de dados incorretos/defeituosos no nível do registro.

Conteúdo relacionado

• Visão geral das práticas recomendadas do Fabric Spark
• Capacidade e planejamento de cluster do Fabric Spark
• Segurança do Fabric Spark
• Desenvolvimento e monitoramento