View a markdown version of this page

Removendo partições dinâmicas - AWS Recomendações

As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Removendo partições dinâmicas

Spark3.0 e versões posteriores incluem Dynamic Partition Pruning (DPP). A remoção dinâmica de partições é uma técnica de otimização Spark que evita a digitalização de partições desnecessárias ao ler dados. A seguir estão algumas coisas importantes que você deve saber sobre o DPP:

  • Ele examina os valores de partição solicitados nos filtros e predicados da consulta e determina quais partições são necessárias para satisfazer a consulta. Todas as partições consideradas desnecessárias são removidas de forma automática e transparente.

  • O DPP reduz o tempo de processamento e a utilização de recursos ignorando partições que não contêm dados aplicáveis. Isso ajuda Spark a focar somente nas partições relevantes.

  • Ele funciona com partições estáticas e partições geradas dinamicamente que são adicionadas por meio de inserções ou cargas incrementais. Sparkreconhece novas partições e pode continuar aplicando a poda dinâmica.

  • O DPP é completamente transparente ou invisível para os desenvolvedores. Nenhuma codificação especial é necessária para habilitar o DPP. Ela ocorre automaticamente nos bastidores como uma otimização durante a geração do plano de consulta.

A seguir estão algumas das melhores práticas para garantir que o DPP funcione com eficiência:

  • Use a redução de predicados aplicando filtros logo no início de suas operações de Spark data frame. Isso ajuda a Spark eliminar partições precocemente usando os metadados da partição.

  • Colete estatísticas sobre seus dados executando ANALYZE TABLE com frequência. Isso reduz as estatísticas em nível de coluna que ajudam a Spark determinar com mais precisão quais partições podem ser ignoradas.

  • Evite particionar demais seus dados. Muitas partições podem sobrecarregar o nó do driver quando ele coleta estatísticas. Escolha entre 10 e 100 partições para cada mesa grande.

  • Reparticione os quadros de dados antes das uniões. Isso evita junções aleatórias que exijam a movimentação de todos os dados e otimiza ainda mais a quantidade de dados lidos.

  • Use tipos de colunas de partição e nomenclatura consistentes nas diferentes tabelas que estão sendo unidas. Isso ajuda a combinar Spark melhor as partições para otimização de junções.

  • Teste as consultas EXPLAIN para garantir que o DPP esteja sendo aplicado e verifique se o ajuste adicional é necessário.

Em um esquema em estrela, as tabelas são divididas em dois tipos principais: tabelas de fatos e tabelas de dimensões. As tabelas de dimensões tendem a ser muito menores do que as tabelas de fatos. Ao unir uma tabela de fatos a uma tabela de dimensões, o DPP otimiza o plano de consulta. Ele cria uma subconsulta a partir de qualquer filtro aplicado à tabela de dimensões. Ele transmite essa subconsulta e cria uma tabela de hash a partir dela. Em seguida, ele aplica a tabela de hash à fase de verificação da tabela de fatos, antes de ler os dados da tabela de fatos. Isso ajuda o DPP a reduzir a quantidade de dados que devem ser lidos na tabela de fatos maior.

O exemplo de consulta a seguir mostra o DPP em ação. A consulta busca o número de pedidos do país (Índia) e inclui uma junção interna entre uma tabela de fatos (fact_orders) e uma tabela de dimensões (nation). A fact_orders tabela é particionada pela colunao_nationkey.

- "select n.n_name as country, count(1) as no_of_orders from fact_orders o join nation n on o.o_nationkey = n.n_nationkey where n.n_name = 'INDIA' group by n.n_name"

A seguir estão as etapas usadas no EXPLAIN plano:

  1. Examine a tabela de dimensões menores (nation) e filtre por colunan_name = 'INDIA'.

  2. Transmita os resultados da etapa anterior.

  3. Crie uma subconsulta que filtre os resultados da primeira etapa.

  4. Empurre-o para baixo PartitionFilter para que ele verifique somente as partições da tabela de fatos que são necessárias, em vez de uma varredura completa da tabela.

A seguir está o EXPLAIN plano para essa consulta otimizada para DPP.

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   *(4) HashAggregate(keys=[], functions=[count(1)], output=[count#208L])
   +- ShuffleQueryStage 3
      +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [id=#353]
         +- *(3) HashAggregate(keys=[], functions=[partial_count(1)], output=[count#212L])
            +- *(3) HashAggregate(keys=[n_name#31], functions=[], output=[])
               +- ShuffleQueryStage 1
                  +- Exchange hashpartitioning(n_name#31, 36), ENSURE_REQUIREMENTS, [id=#315]
                     +- *(2) HashAggregate(keys=[n_name#31], functions=[], output=[n_name#31])
                        +- *(2) Project [n_name#31]
                           +- *(2) BroadcastHashJoin [cast(o_nationkey#145 as bigint)], [n_nationkey#32L], Inner, BuildRight, false
                              :- *(2) ColumnarToRow
                              :  +- FileScan parquet [o_nationkey#145] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[s3://aws-spark-tuning/fact_orders], PartitionFilters: [isnotnull(o_nationkey#145), dynamicpruningexpression(cast(o_nationkey#145 as bigint) IN dynamicp..., PushedFilters: [], ReadSchema: struct<>
                              :        +- SubqueryBroadcast dynamicpruning#210, 0, [n_nationkey#32L], [id=#200]
                              :           +- OutputAdapter [n_name#31, n_nationkey#32L]
                              :              +- AdaptiveSparkPlan isFinalPlan=true
                              :                 +- BroadcastQueryStage 2
                              :                    +- ReusedExchange [n_name#31, n_nationkey#32L], BroadcastExchange HashedRelationBroadcastMode(List(input[1, bigint, false]),false), [id=#233]
                              +- BroadcastQueryStage 0
                                 +- BroadcastExchange HashedRelationBroadcastMode(List(input[1, bigint, false]),false), [id=#233]
                                    +- *(1) Filter ((isnotnull(n_name#31) AND (n_name#31 = INDIA)) AND isnotnull(n_nationkey#32L))
                                       +- FileScan json [n_name#31,n_nationkey#32L] Batched: false, DataFilters: [isnotnull(n_name#31), (n_name#31 = INDIA), isnotnull(n_nationkey#32L)], Format: JSON, Location: InMemoryFileIndex[s3://aws-spark-tuning/input/demo/json/nation], PartitionFilters: [], PushedFilters: [IsNotNull(n_name), EqualTo(n_name,INDIA), IsNotNull(n_nationkey)], ReadSchema: struct<n_name:string,n_nationkey:bigint>

Embora não haja nenhum filtro direto adicionado à o_nationkey coluna, devido ao recurso DPP, verifica Spark automaticamente somente as partições necessárias, em vez de toda a tabela.