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

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

1. 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).

   

2. 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()}")
   

3. 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.

Conteúdo relacionado

• Visão geral das práticas recomendadas do Fabric Spark
• Capacidade de Spark no tecido e planeamento de clusters
• Fabric Spark Security
• Desenvolvimento e Monitorização