Otimizar as operações de shuffle

Certas operações, como join() egroupByKey(), exigem que o Spark execute uma operação de shuffle. O shuffle é o mecanismo do Spark para redistribuir dados para que sejam agrupados de forma diferente nas partições do RDD. Realizar a operação de shuffle pode ajudar a remediar gargalos de performance. No entanto, como o a operação de shuffle geralmente envolve a cópia de dados entre os executores do Spark, o shuffle é uma operação complexa e cara. Por exemplo, os shuffles geram os seguintes custos:

O shuffle é um dos fatores mais importantes na degradação da performance da sua aplicação do Spark. Ao armazenar os dados intermediários, ele pode esgotar espaço no disco local do executor, o que faz com que a tarefa do Spark falhe.

Você pode avaliar a performance do shuffle nas métricas do CloudWatch e na interface do usuário do Spark.

métricas do CloudWatch

Se o valor de Bytes gravados de shuffle for alto em comparação com o valor de Bytes lidos de shuffle, seu trabalho do Spark poderá usar operações de shuffle, como ou join() ou groupByKey().

IU do Spark

Na guia Etapa da interface do usuário do Spark, você pode verificar os valores de Registros/Tamanho de gravações de shuffle. Você também pode verificá-los na guia Executores.

Na captura de tela a seguir, cada executor troca aproximadamente 18,6 GB/4.020.000 registros com o processo shuffle, para um tamanho total de leitura de shuffle de cerca de 75 GB.

A coluna Vazamento de shuffle (Disco) mostra um grande volume de derramamento de dados da memória para o disco, o que pode causar um disco cheio ou um problema de performance.

Se você observar esses sintomas e a etapa demorar muito em comparação com suas metas de performance, ou se ela falhar com os erros Out Of Memory ou No space left on device, considere as soluções a seguir.

Otimizar a junção

A operação join(), que faz a junção de tabelas, é a operação de shuffle mais usada, mas geralmente é um gargalo de performance. Como a junção é uma operação onerosa, recomendamos não usá-la, a menos que seja essencial para seus requisitos de negócios. Certifique-se de que você está utilizando o seu pipeline de dados de forma eficiente, questionando-se sobre o seguinte:

Depois de confirmar que suas operações de junção são essenciais para seus requisitos de negócios, consulte as opções a seguir para otimizar sua junção de forma que atenda às suas necessidades.

Usar o pushdown antes realizar uma junção

Filtre linhas e colunas desnecessárias no DataFrame antes de realizar uma junção. Essa abordagem tem as seguintes vantagens:

# Default
df_joined  = df1.join(df2, ["product_id"])

# Use Pushdown
df1_select = df1.select("product_id","product_title","star_rating").filter(col("star_rating")>=4.0)
df2_select = df2.select("product_id","category_id")
df_joined  = df1_select.join(df2_select, ["product_id"])

Usar junção de DataFrame

Tente usar uma API de alto nível do Spark, como SparkSQL, DataFrame e Datasets, em vez da API do RDD ou da junção de DynamicFrame. Você pode converter o DynamicFrame em DataFrame com uma chamada de método como dyf.toDF(). Conforme analisado na seção Principais tópicos do Apache Spark, essas operações de junção aproveitam internamente a otimização de consultas pelo otimizador Catalyst.

Dicas e junções shuffle e broadcast hash

O Spark é compatível com dois tipos de junção: junção shuffle e junção broadcast hash. Uma junção broadcast hash não requer operação de shuffle e pode exigir menos processamento do que uma junção shuffle. No entanto, é aplicável somente ao realizar a junção de uma tabela pequena com uma grande. Ao realizar a junção de uma tabela que cabe na memória de um único executor do Spark, considere usar uma junção broadcast hash.

O diagrama a seguir mostra a estrutura de alto nível e as etapas de uma junção broadcast hash e uma junção shuffle.

Os detalhes de cada junção são os seguintes:

No AWS Glue versão 3.0 e posterior, você pode aproveitar automaticamente as junções broadcast hash habilitando a Execução Adaptativa de Consultas e parâmetros adicionais. A Execução Adaptativa de Consultas converte uma junção sort-merge em uma junção broadcast hash quando as estatísticas de runtime de um dos lados da junção são menores do que o limite de junção broadcast hash adaptativa.

No AWS Glue versão 3.0, você pode habilitar a Execução Adaptativa de Consultas definindo spark.sql.adaptive.enabled=true. A Execução Adaptativa de Consultas está habilitada por padrão no AWS Glue 4.0.

Você pode definir parâmetros adicionais relacionados aos shuffles e às junções broadcast hash:

Para obter mais informações sobre parâmetros relacionados, consulte Converting sort-merge join to broadcast join.

No AWS Glue versão 3.0 ou posterior, você pode usar outras dicas de junção para shuffle para ajustar seu comportamento.

-- Join Hints for shuffle sort merge join
SELECT /*+ SHUFFLE_MERGE(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;
SELECT /*+ MERGEJOIN(t2) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;
SELECT /*+ MERGE(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

-- Join Hints for shuffle hash join
SELECT /*+ SHUFFLE_HASH(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

-- Join Hints for shuffle-and-replicate nested loop join
SELECT /*+ SHUFFLE_REPLICATE_NL(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

Usar bucketing

A junção sort-merge requer duas fases: shuffle e classificar, e depois mesclar. Essas duas fases podem sobrecarregar o executor do Spark e causar problemas de OOM e performance quando alguns dos executores estão sendo mesclados e outros estão sendo classificados simultaneamente. Nesses casos, talvez seja possível juntar com eficiência usando bucketing. O bucketing realizará o pré-shuffle e pré-classificará sua entrada nas chaves de junção. Em seguida, gravará esses dados classificados em uma tabela intermediária. O custo das etapas de shuffle e classificação pode ser reduzido ao juntar tabelas grandes definindo as tabelas intermediárias classificadas com antecedência.

As tabelas em buckets são úteis para o seguinte:

Você pode criar uma tabela em buckets usando o código a seguir.

df.write.bucketBy(50, "account_id").sortBy("age").saveAsTable("bucketed_table")

Reparticionar DataFrames nas chaves de junção antes da junção

Para reparticionar os dois DataFrames nas chaves de junção antes da junção, use as instruções a seguir.

df1_repartitioned = df1.repartition(N,"join_key")
df2_repartitioned = df2.repartition(N,"join_key")
df_joined  = df1_repartitioned.join(df2_repartitioned,"product_id")

Isso particionará dois RDDs (ainda separados) na chave de junção antes de iniciar a junção. Se os dois RDDs forem particionados na mesma chave com o mesmo código de particionamento, o RDD registrará que seu plano de juntar terá uma grande probabilidade de coexistir no mesmo operador antes de realizar o shuffle para a junção. Isso pode melhorar a performance ao reduzir a atividade da rede e a distorção de dados durante a junção.

Superar distorções de dados

A distorção de dados é uma das causas mais comuns de um gargalo nos trabalhos do Spark. Ela ocorre quando os dados não são distribuídos uniformemente nas partições do RDD. Isso faz com que as tarefas dessa partição demorem muito mais do que outras, atrasando o tempo geral de processamento da aplicação.

Para identificar a distorção de dados, avalie as seguintes métricas na interface do usuário do Spark:

Se você vir distorções de dados, tente o seguinte:

Usar cache

Ao usar DataFrames repetitivos, evite shuffle ou computação adicional usando df.cache() ou df.persist() para armazenar em cache os resultados do cálculo na memória e no disco de cada executor do Spark. O Spark também é compatível com RDDs persistentes em disco ou replicação em vários nós (nível de armazenamento).

Por exemplo, você pode persistir os DataFrames adicionando df.persist(). Quando o cache não for mais necessário, você poderá usar unpersist para descartar os dados em cache.

df = spark.read.parquet("s3://<Bucket>/parquet/product_category=Books/")
df_high_rate = df.filter(col("star_rating")>=4.0)
df_high_rate.persist()

df_joined1 = df_high_rate.join(<Table1>, ["key"])
df_joined2 = df_high_rate.join(<Table2>, ["key"])
df_joined3 = df_high_rate.join(<Table3>, ["key"])
... 
df_high_rate.unpersist()

Remover ações desnecessárias do Spark

Evite executar ações desnecessárias, como count, show ou collect. Conforme analisado na seção Principais tópicos do Apache Spark, o Spark tem carregamento lento. Cada RDD transformado pode ser recalculado toda vez que você executa uma ação nele. Quando você usa várias ações do Spark, vários acessos à origem, cálculos de tarefas e operações de shuffle são executados para cada ação sendo chamada.

Se você não precisar de collect() ou de outras ações em seu ambiente comercial, considere removê-las.