Boas Práticas de Uso do Spark
1. Uso de UDFs (User Defined Functions)
Recomendação: Evite o uso excessivo de UDFs. Eles podem ser menos otimizados do que as funções nativas do Spark e, assim, reduzir o desempenho.
2. Compreensão do Lazy Evaluation
O Spark utiliza um conceito chamado "Lazy Evaluation", ou seja, ele não executa uma operação até que seja absolutamente necessário. Recomendação: Esteja ciente dessa característica para evitar surpresas durante a execução.
3. Uso de Cache
Utilize .cache() para armazenar um objeto (como um DataFrame) na memória. Isso pode acelerar operações subsequentes no mesmo objeto. Use .uncache() para remover um objeto previamente armazenado em cache da memória. Recomendação: Use o cache sabiamente. Não cache todos os DataFrames indiscriminadamente. Avalie quais objetos realmente beneficiariam do cache.
4. Configurações para Otimização de Memória
Ative spark.sql.inMemoryColumnarStorage.compressed para armazenamento colunar comprimido em memória. Ao armazenar DataFrames em cache, prefira o formato colunar para otimizar o espaço e a velocidade de acesso.
5. Otimizações de Join
Use "Broadcast Join" quando uma das tabelas em sua operação join for pequena o suficiente para caber na memória. A configuração spark.sql.autoBroadcastJoinThreshold determina o tamanho máximo de uma tabela que pode ser transmitida para todos os nós durante um join. Ajuste conforme necessário. Dica: [Broadcast Hash Join with Spark DataFrame](https://stackoverflow.com/questions/47744265/broadcast-hash-join-with-spark-dataframe)
6. Ativação de Otimizadores
Ative o Otimizador Baseado em Custo (CBO) usando spark.sql.cbo.enabled. Ative o Adaptive Query Execution (AQE) com spark.sql.adaptive.enabled.
Recomendação: Evite o uso excessivo de UDFs. Eles podem ser menos otimizados do que as funções nativas do Spark e, assim, reduzir o desempenho.
2. Compreensão do Lazy Evaluation
O Spark utiliza um conceito chamado "Lazy Evaluation", ou seja, ele não executa uma operação até que seja absolutamente necessário. Recomendação: Esteja ciente dessa característica para evitar surpresas durante a execução.
3. Uso de Cache
Utilize .cache() para armazenar um objeto (como um DataFrame) na memória. Isso pode acelerar operações subsequentes no mesmo objeto. Use .uncache() para remover um objeto previamente armazenado em cache da memória. Recomendação: Use o cache sabiamente. Não cache todos os DataFrames indiscriminadamente. Avalie quais objetos realmente beneficiariam do cache.
4. Configurações para Otimização de Memória
Ative spark.sql.inMemoryColumnarStorage.compressed para armazenamento colunar comprimido em memória. Ao armazenar DataFrames em cache, prefira o formato colunar para otimizar o espaço e a velocidade de acesso.
5. Otimizações de Join
Use "Broadcast Join" quando uma das tabelas em sua operação join for pequena o suficiente para caber na memória. A configuração spark.sql.autoBroadcastJoinThreshold determina o tamanho máximo de uma tabela que pode ser transmitida para todos os nós durante um join. Ajuste conforme necessário. Dica: [Broadcast Hash Join with Spark DataFrame](https://stackoverflow.com/questions/47744265/broadcast-hash-join-with-spark-dataframe)
6. Ativação de Otimizadores
Ative o Otimizador Baseado em Custo (CBO) usando spark.sql.cbo.enabled. Ative o Adaptive Query Execution (AQE) com spark.sql.adaptive.enabled.
7. Gerenciamento de Partições
Equilíbrio de Partições: Poucas partições podem resultar em alguns executores ociosos, enquanto muitas partições podem levar à sobrecarga de agendamento de tarefas. Encontrar o equilíbrio certo é essencial. Configuração de Partições: Defina a quantidade e o tamanho ideal das partições do seu RDD de acordo com o contexto do seu trabalho. Utilize a configuração spark.sql.shuffle.partitions (valor padrão é 200) para ajustar conforme necessário. Redução de Embaralhamento: O embaralhamento de dados pode ser uma operação custosa. Tente minimizá-la sempre que possível. Filtre Dados: Antes de realizar operações, filtre seus dados para trabalhar apenas com as informações necessárias.
8. Evite Operações Custosas
Ordenação: Evite usar order by a menos que seja estritamente necessário. Seleção de Colunas: Em vez de usar select * para recuperar todas as colunas, especifique e recupere apenas as colunas relevantes para sua consulta.
9. Uso Eficiente da Repartição e Coalesce
Se desejar reduzir o número de partições, prefira usar coalesce(). É uma versão otimizada em comparação com repartition(). O coalesce é mais eficiente, especialmente com grandes conjuntos de dados, pois move menos dados entre partições.
10. Minimize operações com Shuffle e configure de paralelismo
O shuffle pode ser uma operação intensiva e pode afetar o desempenho. Portanto, tente evitar muitas operações que envolvam shuffle. spark.default.parallelism: Define o número padrão de partições em operações de RDD sem shuffle, como parallelize, map e filter. O padrão varia entre tipos de cluster. No local mode, o padrão é o número de núcleos disponíveis; em um cluster YARN, o padrão é o número máximo de tarefas simultâneas que Spark pode executar em todo o cluster.
11. Configuração da Memória Off-Heap
A memória off-heap refere-se à alocação de memória fora da heap do Java, permitindo que o Spark use memória além da memória máxima configurada para a JVM. Ativação da Memória Off-Heap: spark.memory.offHeap.enabled: Ativa o uso da memória off-heap. Por padrão, é false. Para ativar, defina como true. spark.memory.offHeap.size: Define o tamanho da memória off-heap. Deve ser especificado em bytes, ou você pode usar sufixos padrão como "k", "m", "g", etc. (por exemplo, "512m").
Equilíbrio de Partições: Poucas partições podem resultar em alguns executores ociosos, enquanto muitas partições podem levar à sobrecarga de agendamento de tarefas. Encontrar o equilíbrio certo é essencial. Configuração de Partições: Defina a quantidade e o tamanho ideal das partições do seu RDD de acordo com o contexto do seu trabalho. Utilize a configuração spark.sql.shuffle.partitions (valor padrão é 200) para ajustar conforme necessário. Redução de Embaralhamento: O embaralhamento de dados pode ser uma operação custosa. Tente minimizá-la sempre que possível. Filtre Dados: Antes de realizar operações, filtre seus dados para trabalhar apenas com as informações necessárias.
8. Evite Operações Custosas
Ordenação: Evite usar order by a menos que seja estritamente necessário. Seleção de Colunas: Em vez de usar select * para recuperar todas as colunas, especifique e recupere apenas as colunas relevantes para sua consulta.
9. Uso Eficiente da Repartição e Coalesce
Se desejar reduzir o número de partições, prefira usar coalesce(). É uma versão otimizada em comparação com repartition(). O coalesce é mais eficiente, especialmente com grandes conjuntos de dados, pois move menos dados entre partições.
10. Minimize operações com Shuffle e configure de paralelismo
O shuffle pode ser uma operação intensiva e pode afetar o desempenho. Portanto, tente evitar muitas operações que envolvam shuffle. spark.default.parallelism: Define o número padrão de partições em operações de RDD sem shuffle, como parallelize, map e filter. O padrão varia entre tipos de cluster. No local mode, o padrão é o número de núcleos disponíveis; em um cluster YARN, o padrão é o número máximo de tarefas simultâneas que Spark pode executar em todo o cluster.
11. Configuração da Memória Off-Heap
A memória off-heap refere-se à alocação de memória fora da heap do Java, permitindo que o Spark use memória além da memória máxima configurada para a JVM. Ativação da Memória Off-Heap: spark.memory.offHeap.enabled: Ativa o uso da memória off-heap. Por padrão, é false. Para ativar, defina como true. spark.memory.offHeap.size: Define o tamanho da memória off-heap. Deve ser especificado em bytes, ou você pode usar sufixos padrão como "k", "m", "g", etc. (por exemplo, "512m").
12. Modificar o Compression Block Size
A compressão pode ajudar a reduzir a quantidade de dados sendo transmitidos entre as fases e, consequentemente, acelerar o embaralhamento.
• Compression Block Size:
• spark.io.compression.lz4.blockSize: Define o tamanho do bloco para a compressão LZ4, quando essa compressão está sendo usada. O tamanho padrão é 32k.
13. Broadcasting
Broadcasting é uma técnica usada no Spark para otimizar operações de join quando um dos DataFrames (ou RDDs) é pequeno o suficiente para ser enviado para todos os nós do cluster.
• Usando Broadcasting: Carregue um pequeno conjunto de dados na memória para juntar-se a um grande conjunto de dados. Isso evita o embaralhamento e pode acelerar significativamente as operações de join.
A compressão pode ajudar a reduzir a quantidade de dados sendo transmitidos entre as fases e, consequentemente, acelerar o embaralhamento.
• Compression Block Size:
• spark.io.compression.lz4.blockSize: Define o tamanho do bloco para a compressão LZ4, quando essa compressão está sendo usada. O tamanho padrão é 32k.
13. Broadcasting
Broadcasting é uma técnica usada no Spark para otimizar operações de join quando um dos DataFrames (ou RDDs) é pequeno o suficiente para ser enviado para todos os nós do cluster.
• Usando Broadcasting: Carregue um pequeno conjunto de dados na memória para juntar-se a um grande conjunto de dados. Isso evita o embaralhamento e pode acelerar significativamente as operações de join.
Referências
Spark Shuffle Partitions vs Default Parallelism
Towards Data Science - Apache Spark Configuration
Data for Geeks - Apache Spark Performance Tuning
Apache Spark Optimization Techniques by Ch-Nabarun
Analytics Vidhya - Apache Spark Memory Management
Dive into Spark Memory by Luminousmen
Official Spark Configuration Documentation
Spark by Examples - Spark Performance Tuning
Spark by Examples - Spark SQL Performance Tuning