Sort no Spark

Quando falamos em sort no Spark, falamos em ordenar os dados dentro dos nós antes de realizar uma segunda ordenação para o bucket de destino.

Os dados são gerenciados em memória nos nós, mas nos casos onde os datasets são grandes demais para ficarem em memória, ocorrem spills para o disco de forma intermediária, para que essa ordenação seja realizada da melhor forma.

Diagrama

flowchart TD
    subgraph Stage1["Estágio 1"]
        E1["Executor 1\n[5, 2, 8, 1, 9]"]
        E2["Executor 2\n[3, 7, 4, 6, 0]"]
    end

    RP(["RangePartitioner\namostra os dados e calcula\nfronteiras: [3, 6]"])

    subgraph SW["Shuffle Write (disco local)"]
        W1B0["E1: bucket 0\n[2, 1]"]
        W1B1["E1: bucket 1\n[5]"]
        W1B2["E1: bucket 2\n[8, 9]"]
        W2B0["E2: bucket 0\n[0]"]
        W2B1["E2: bucket 1\n[3, 4]"]
        W2B2["E2: bucket 2\n[7, 6]"]
    end

    subgraph SR["Shuffle Read (rede)"]
        R0["Reducer 0\nbucket 0 de todos"]
        R1["Reducer 1\nbucket 1 de todos"]
        R2["Reducer 2\nbucket 2 de todos"]
    end

    subgraph Stage2["Estágio 2 (sort local por partição)"]
        O0["Partição 0\n[0, 1, 2]"]
        O1["Partição 1\n[3, 4, 5]"]
        O2["Partição 2\n[6, 7, 8, 9]"]
    end

    E1 & E2 -->|amostragem| RP
    RP -->|"chave < 3"| W1B0 & W2B0
    RP -->|"3 ≤ chave < 6"| W1B1 & W2B1
    RP -->|"chave ≥ 6"| W1B2 & W2B2
    W1B0 & W2B0 --> R0
    W1B1 & W2B1 --> R1
    W1B2 & W2B2 --> R2
    R0 --> O0
    R1 --> O1
    R2 --> O2

A diferença central em relação ao spark-shuffle comum: o RangePartitioner substitui o HashPartitioner. Antes de redistribuir os dados, o Spark amostra o dataset para estimar a distribuição das chaves e calcular fronteiras que dividem o espaço em faixas de tamanho parecido. Com isso, cada reducer recebe um intervalo contíguo de valores e só precisa fazer um sort local no final. O resultado é ordenação global: tudo na partição 0 é menor que qualquer valor na partição 1, e assim por diante.

Conexões

• spark-shuffle: operação base do sort; o sort adiciona RangePartitioner e ordenação local
• spark-arquitetura: onde o sort se encaixa no modelo de Stages e Tasks
• spark-tamanho-blocos: tamanho das partições afeta o custo do sort
• sistemas-distribuidos-particionamento: estratégias de particionamento incluindo range partitioning

Referências

• Spark SQL ordenação