Consultar tabelas distribuídas no Azure Cosmos DB para PostgreSQL

  • 7 minutos

O Woodgrove Bank pediu que você otimizasse o desempenho das cargas de trabalho e consultas de análise de usuários do aplicativo de pagamento sem contato. Aplicando as informações aprendidas na Unidade 6, você está pronto para examinar como executar consultas em dados distribuídos.

Normalmente, você emite consultas usando consultas PostgreSQL padrão no nó coordenador no Azure Cosmos DB para PostgreSQL SELECT . Ele lida com consultas de paralelização SELECT envolvendo seleções, agrupamentos, pedidos e JOINs complexos para otimizar e acelerar o desempenho da consulta.

Consultar tabelas distribuídas

Em um alto nível, o coordenador particiona cada SELECT consulta em fragmentos de consulta menores, atribui fragmentos de consulta aos trabalhadores, supervisiona sua execução, mescla seus resultados e retorna o resultado final ao usuário. Na maioria das vezes, não é necessário saber como ou onde os dados são armazenados em um cluster. O banco de dados usa um executor de consulta distribuído para dividir automaticamente consultas SQL regulares e executá-las em paralelo em nós de trabalho próximos aos dados.

Para consultas simples, como SELECT COUNT(*) FROM payment_users;, o coordenador executa automaticamente a contagem de todos os fragmentos em paralelo e combina os resultados.

Executar funções agregadas

Para as agregações e consultas mais complexas associadas às cargas de trabalho analíticas do Woodgrove Bank, o Azure Cosmos DB para PostgreSQL dá suporte e paraleliza a maioria das funções agregadas suportadas pelo PostgreSQL, incluindo agregações personalizadas definidas pelo usuário.

Animation of the flow of a query arriving from an application is displayed.

Os agregados são executados usando um dos três métodos, nesta ordem de preferência:

  1. Quando a agregação é agrupada pela coluna de distribuição de uma tabela, o coordenador pode empurrar para baixo a execução de toda a consulta para cada trabalhador. Todos os agregados são suportados nessa situação e executados em paralelo nos nós de trabalho. Por exemplo, você pode contar o número de eventos, por tipo, por usuário no aplicativo de pagamentos sem contato usando a seguinte consulta:

    SELECT user_id, event_type, count(*) FROM payment_events GROUP By user_id, event_type;

    Essa consulta é executada rapidamente porque o coordenador pode enviar a execução da consulta para cada nó de trabalho. A execução push-down é possível porque a cláusula contém a GROUP BY coluna de distribuição da tabela. Executar a mesma consulta usando a instrução permite que você veja o plano de execução da consulta e como ele é distribuído entre os nós de trabalho para paralelizar a EXPLAIN VERBOSE execução da consulta.

  2. Quando a agregação não é agrupada pela coluna de distribuição de uma tabela, o coordenador ainda pode otimizar caso a caso. Regras internas para agregados específicos como sum(), avg()e count(distinct) permitem que as consultas sejam reescritas para agregação parcial em trabalhadores. Por exemplo, para calcular uma média, o coordenador obtém uma soma e uma contagem de cada trabalhador e, em seguida, o nó coordenador calcula a média final.

    SELECT merchant_id, event_type, COUNT(*) FROM payment_events GROUP BY merchant_id, event_type;

    Essa consulta resulta em uma agregação parcial ocorrendo em trabalhadores, que é um pouco menos eficiente do que o método anterior. A EXPLAIN VERBOSE declaração fornece os detalhes do realizado para recuperar a HashAggregate contagem de cada trabalhador. Em seguida, o coordenador calcula a contagem final.

  3. Para todas as outras funções de agregação, o coordenador puxa todas as linhas dos trabalhadores e executa a agregação em si. Se os dois métodos anteriores não cobrirem o agregado, o coordenador recorre a esta abordagem. No entanto, é fundamental observar que esse método pode causar sobrecarga de rede e esgotar os recursos do coordenador se o conjunto de dados a ser agregado for muito grande.

Associações

As junções entre qualquer número de tabelas, independentemente do seu tamanho e método de distribuição, são suportadas. O planejador de consultas seleciona o método de junção ideal e a ordem com base em como as tabelas são distribuídas. Ele avalia várias ordens de junção possíveis e cria um plano de junção que requer a quantidade mínima de dados a serem transferidos pela rede.

Dadas as pesadas cargas de trabalho analíticas que o Woodgrove Bank executa em relação ao banco de dados, você precisará oferecer suporte e realizar junções em várias tabelas. Quando o banco de dados é construído, o Woodgrove terá mais do que apenas as tabelas de usuários, eventos e comerciantes, algumas das quais podem ser colocalizadas e outras não.

Junta-se co-localizada

Quando duas tabelas são colocalizadas, elas podem ser unidas de forma eficiente em suas colunas de distribuição comuns. Uma junção colocalizada é a maneira mais eficiente de unir duas grandes mesas distribuídas.

O coordenador usa suas tabelas de metadados para determinar quais fragmentos das tabelas colocalizadas podem corresponder aos fragmentos da outra tabela. Esse processo permite que o coordenador elimine pares de estilhaços que não podem produzir chaves de junção correspondentes. As junções entre os pares de estilhaços restantes são executadas em paralelo nos nós de trabalho e os resultados são devolvidos ao coordenador.

Nota

Certifique-se de que as tabelas estão distribuídas no mesmo número de fragmentos e que as colunas de distribuição de cada tabela têm tipos exatamente correspondentes. Tentar unir em colunas de tipos ligeiramente diferentes, como int e bigint pode causar problemas.

Para o aplicativo de pagamento sem contato, você colocou os payment_events dados e payment_users usando a mesma coluna de distribuição, user_id. A execução de uma junção entre essas duas tabelas na user_id coluna permite que o coordenador use as tabelas de metadados para determinar eficientemente as linhas com chaves correspondentes e paralelizar a execução da consulta entre nós de trabalho.

SELECT u.user_id, login, event_type, merchant_id, event_details FROM payment_events e INNER JOIN payment_users u ON e.user_id = u.user_id LIMIT 5;

Associações de repartição

Às vezes, talvez seja necessário unir duas tabelas em colunas diferentes da coluna de distribuição. Para esses casos, o Azure Cosmos DB para PostgreSQL permite unir colunas de chave que não sejam de distribuição reparticionando dinamicamente as tabelas para a consulta. As junções de repartição exigem embaralhamento cruzado de dados, por isso são menos eficientes do que as junções colocalizadas. Seria melhor se você tentasse distribuir suas tabelas por chaves de junção comuns sempre que possível.

Nesses casos, o otimizador de consulta determina a(s) tabela(s) a ser particionada com base nas colunas de distribuição, chaves de junção e tamanhos. Com tabelas reparticionadas, apenas pares de estilhaços relevantes são unidos entre si, reduzindo drasticamente a quantidade de dados transferidos pela rede.

Há inúmeras consultas que o Woodgrove Bank usa em suas cargas de trabalho de análise que exigirão consultas de repartição, portanto, você precisa configurar o banco de dados para permitir esses tipos de consultas. Você pode habilitar a execução de consultas de repartição executando o seguinte comando:

SET citus.enable_repartition_joins TO on;

Suponha que as junções de repartição não tenham sido habilitadas no banco de dados. A execução de uma junção que requer reparticionamento resulta na mensagem de erro: ERROR: the query contains a join that requires repartitioning.

Suponha por um momento que você não converteu a tabela do payment_merchants Woodgrove Bank em uma tabela de referência, deixando-a distribuída merchant_id na coluna. A junção das payment_merchants tabelas e payment_events exigiria uma junção de repartição. Depois de habilitar as junções de repartição, você pode executar a seguinte junção em uma coluna que não seja de distribuição:

SELECT m.merchant_id, name, event_type, count(*) as event_count
FROM payment_events e
JOIN payment_merchants m ON e.merchant_id = m.merchant_id
WHERE event_type = 'SendFunds'
GROUP BY m.merchant_id, name, event_type
ORDER BY event_count DESC
LIMIT 5;

Juntas à tabela de referência

Você pode usar tabelas de referência como tabelas de "dimensão" para unir eficientemente com grandes tabelas de "fatos". As tabelas de referência são replicadas em todos os nós de trabalho, permitindo que uma junção de referência seja decomposta em junções locais em cada trabalhador e executada em paralelo. Uma junção de referência é como uma versão mais flexível de uma junção colocalizada porque as tabelas de referência não são distribuídas em nenhuma coluna específica e são livres para ingressar em qualquer uma de suas colunas.

Para preencher um painel para o aplicativo de pagamento sem contato do Woodgrove Bank, você foi solicitado a escrever uma consulta para contar o número de transações por tipo para cada comerciante. Esta consulta requer a junção da tabela distribuída com a payment_eventspayment_merchants tabela de referência na merchant_id coluna.

SELECT m.merchant_id, name, event_type, count(*) as event_count
FROM payment_events e
JOIN payment_merchants m ON e.merchant_id = m.merchant_id
WHERE event_type = 'SendFunds'
GROUP BY m.merchant_id, name, event_type
ORDER BY event_count DESC
LIMIT 5;

Executar a consulta com EXPLAIN VERBOSE mostra como o coordenador pode gerar um plano que envia a execução da consulta para cada um dos 32 fragmentos, onde a tabela de referência é unida localmente nos nós de trabalho. Nesse caso, alterar a tabela para uma tabela de referência fornece melhorias significativas de desempenho em relação à mesma consulta executada em relação a payment_merchants uma tabela distribuída não colocalizada.

As tabelas de referência também podem se unir com tabelas locais no nó coordenador.

Modificar dados em tabelas distribuídas

A execução UPDATE e os comandos em tabelas distribuídas são realizadas usando o PostgreSQL UPDATE padrão e DELETEDELETE comandos. Eles podem ser concluídos sem especificar a coluna de distribuição em uma WHERE cláusula, mas serão executados de forma mais eficiente se ela for incluída.

Atualizar linhas em uma tabela distribuída

Use o comando PostgreSQL UPDATE padrão para atualizar registros armazenados em tabelas distribuídas. Por exemplo, o Woodgrove Bank pediu que você modificasse o campo de cada registro no banco de dados para incluir o event_detailsuser_id valor na cadeia de caracteres JSONB.

UPDATE payment_events
SET event_details = jsonb_set(event_details, '{user_id}', CAST(user_id as text)::jsonb);

Quando as atualizações afetam vários fragmentos, como no exemplo anterior, o comportamento padrão é usar um protocolo de confirmação de fase única. Esse comportamento significa que cada trabalhador envia uma mensagem "concluído" para o coordenador e, em seguida, aguarda uma mensagem de confirmação ou anulação do coordenador. Quando todos os trabalhadores terminarem de executar a consulta e enviarem uma mensagem "concluída", o coordenador decide se confirma ou aborta a transação.

Para maior segurança, você pode habilitar a confirmação bifásica definindo o citus.multi_shard_commit_protocol valor como , da 2pcseguinte maneira:

SET citus.multi_shard_commit_protocol = '2pc';

Se o afeta apenas um único fragmento, ele é executado dentro de um único nó de trabalho, e habilitar o UPDATE 2PC é desnecessário. Esse cenário geralmente ocorre quando atualiza ou exclui o filtro pela coluna de distribuição de uma tabela, como na consulta a seguir:

UPDATE payment_events
SET event_details = jsonb_set(event_details, '{user_id}', CAST(user_id as text)::jsonb)
WHERE user_id = 796958;

Excluir registros de uma tabela distribuída

A exclusão de linhas de tabelas distribuídas também usa o comando PostgreSQL DELETE padrão. No aplicativo de pagamentos, por exemplo, o Woodgrove ocasionalmente precisa executar uma operação para excluir registros de transações duplicados causados por um usuário clicar duas vezes no botão de envio de transação. Eles afirmaram que essa tarefa poderia ser realizada excluindo o registro de transação mais recente da tabela, da payment_events seguinte maneira:

DELETE FROM payment_events
WHERE user_id = 796958
AND created_at = (SELECT MAX(created_at) FROM payment_events WHERE user_id = 796958);

Assim como as atualizações, as operações de exclusão usarão um protocolo de confirmação de fase única por padrão.

Maximizando o desempenho de gravação

Você pode consultar a seção Scaling out Data Ingestion da documentação do Citus para saber mais sobre como maximizar o desempenho de gravação.