Removendo partições dinâmicas - AWS Orientação prescritiva

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

Spark 3.0 e versões posteriores incluem Dynamic Partition Pruning (DPP). A poda dinâmica de partições é uma técnica de otimização em Spark isso 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 para se concentrar 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. Spark reconhece 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 pressão de predicados aplicando filtros logo no início de seu Spark operações de data frame. Isso ajuda Spark para 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 Spark para 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 Spark melhor combinação de partições para otimização de junção.

  • 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>

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