MaxCompute:Arquitetura de integração de dados incrementais e completos em tempo quase real com Delta Tables

Criação de tabelas

O MaxCompute introduz as Delta tables com um formato de dados de tabela unificado para suportar a arquitetura integrada. As Delta tables suportam todos os recursos dos workflows de processamento em lote existentes, além de novos workflows, como armazenamento e processamento de dados incrementais em tempo quase real.

Para crie uma Delta table, especifique as chaves primárias e defina "transactional"="true" na instrução CREATE TABLE:

CREATE TABLE tt2 (pk BIGINT NOT NULL PRIMARY KEY, val STRING) tblproperties ("transactional"="true");
CREATE TABLE par_tt2 (pk BIGINT NOT NULL PRIMARY KEY, val STRING)  PARTITIONED BY (pt STRING) tblproperties ("transactional"="true");

As chaves primárias garantem a unicidade das linhas. A propriedade transactional habilita o mecanismo de transação ACID (atomicidade, consistência, isolamento e durabilidade) com isolamento de snapshot para operações de leitura e gravação. Para mais detalhes, consulte Operações de tabela.

Parâmetros principais para Delta tables

Para a referência completa de parâmetros, consulte a seção "Parâmetros para Delta tables" em Operações de tabela.

write.bucket.num

Especifica o número de buckets por tabela particionada ou não particionada e o número de nós de gravação simultâneos. O valor padrão é 16. Valores válidos: (0, 4096].

• Para tabelas particionadas: o valor pode ser alterado e se aplica automaticamente às novas partições.

• Para tabelas não particionadas: o valor não pode ser alterado após a criação.

acid.data.retain.hours

Define o intervalo de tempo de dados históricos disponíveis para consultas time travel. O valor padrão é 24. Valores válidos: [0, 168] (unidade: horas).

• Defina como 0 para desativar consultas time travel e reduzir significativamente os custos de armazenamento de dados históricos.

• Para dados históricos com mais de 168 horas (7 dias), entre em contato com o suporte técnico do MaxCompute.

Evolução de schema

As Delta tables suportam evolução completa de schema, incluindo adição e exclusão de colunas. Ao consultar dados históricos com time travel, o sistema lê os dados com base no schema daquela versão histórica.

O exemplo a seguir adiciona uma coluna:

ALTER TABLE tt2 ADD columns (val2 string);

Para a sintaxe DDL completa, consulte Operações de tabela.

Formatos de dados da tabela

A figura acima mostra a estrutura de dados de uma tabela particionada. Os arquivos de dados são fisicamente isolados por partição (armazenados em diretórios separados), e os dados de cada partição são divididos em buckets. As Delta tables usam dois tipos de arquivos de dados:

Governança e otimização automáticas de dados

Problema: proliferação de arquivos pequenos

As Delta tables suportam importação de dados incrementais em tempo quase real, na casa dos minutos. Em cenários de gravação de alto tráfego com muitos buckets, o número de arquivos pequenos de dados incrementais pode crescer rapidamente, causando excesso de solicitações de acesso, custos elevados e baixa eficiência de I/O. Operações intensivas de UPDATE e DELETE agravam ainda mais esse cenário ao gerar grande quantidade de registros históricos intermediários redundantes.

Solução: quatro serviços de governança automatizados

O engine de armazenamento do MaxCompute governa e otimiza automaticamente os dados armazenados sem necessidade de configuração manual. O engine identifica inteligentemente as características dos dados em múltiplas dimensões e aplica políticas automaticamente.

O auto partial compact limpa apenas registros históricos cuja data de criação ultrapassa a janela de retenção do time travel.

Para cenários que exigem desempenho máximo de consulta, acione uma compactação principal (major compaction) manualmente:

SET odps.merge.task.mode=service;
ALTER TABLE tt2 compact major;

Para mais informações, consulte COMPACTION.

Upserts em tempo quase real na casa dos minutos

Por que usar Delta tables: O processamento em lote tradicional importa dados incrementais para uma nova tabela ou partição ao longo de horas ou dias e, em seguida, aciona um processo ETL offline para unir e mesclar esses dados incrementais com os dados existentes na tabela. Essa abordagem apresenta alta latência e custos elevados de recursos e armazenamento.

Com as Delta tables, o pipeline de upsert mantém uma latência de 5 a 10 minutos entre a gravação dos dados e a consulta. Não há necessidade de processos complexos de mesclagem ETL, o que reduz tanto os custos de computação quanto os de armazenamento.

É comum encontrar diversas fontes de dados em produção, como bancos de dados, sistemas de log e filas de mensagens. O MaxCompute fornece um plug-in conector Flink open-source que funciona com o DataWorks Data Integration e outras ferramentas de importação de dados. Ele suporta design personalizado e otimização de desenvolvimento para cenários de alta concorrência, tolerância a falhas e submissão de transações.

Principais capacidades da integração com o conector Flink:

O throughput de tráfego varia conforme o ambiente e a configuração. Estime o throughput máximo com base na capacidade de processamento de um único bucket (1 MB/s). O grupo de recursos compartilhados do Tunnel é usado por padrão no MaxCompute Tunnel, o que pode causar instabilidade no throughput sob forte contenção de recursos. Também existem limites impostos ao consumo de recursos.

Sincronização de dados em tempo real de bancos de dados usando DataWorks Data Integration

Muitos sistemas de produção combinam processamento de transações online (OLTP), processamento analítico online (OLAP) e engines de análise offline. Um workflow comum é sincronizar novos registros de uma única tabela ou de um banco de dados inteiro para o MaxCompute em tempo real para análise.

A figura acima contrasta duas abordagens:

• Esquerda (processamento em lote): Dados incrementais são importados para uma nova tabela ou partição ao longo de horas ou dias. Um processo ETL offline une e mescla os dados incrementais com os dados existentes na tabela. Essa abordagem apresenta alta latência e custos elevados de recursos e armazenamento.

• Direita (arquitetura integrada): Novos registros são lidos dos bancos de dados em minutos. Não há necessidade de extração ou mesclagem periódica de dados — as Delta tables lidam diretamente com as atualizações, minimizando os custos de computação e armazenamento.

Processamento em lote usando instruções SQL DML e upserts

Os módulos Compiler, Optimizer e Runtime do engine SQL foram modificados e otimizados para operações com Delta tables. Isso inclui análise de sintaxe, planos de otimização, lógica de deduplicação baseada em chave primária e upserts em runtime para fornecer suporte completo à sintaxe SQL.

Comportamentos principais:

• Consistência transacional: Após a conclusão do processamento de dados, o serviço de metadados realiza detecção de conflitos de transação e atualizações atômicas de metadados, garantindo isolamento de leitura/gravação e consistência transacional.

• Upserts simplificados: O sistema mescla automaticamente os registros com base nas chaves primárias durante consultas em Delta tables. Para cenários que misturam operações INSERT e UPDATE, use INSERT INTO em vez de sintaxes complexas como UPDATE ou MERGE INTO. Isso reduz o I/O de leitura e economiza recursos de computação.

Para a sintaxe completa de SQL DML, consulte Operações DML.

Consultas time travel

As consultas time travel permitem consultar versões históricas de uma Delta table. Os casos de uso comuns incluem:

• Recuperação de dados: Restaurar dados para uma versão histórica específica após modificação ou exclusão acidental.

• Rastreamento histórico: Auditar ou reanalisar dados de negócios de um ponto anterior no tempo.

Exemplos de consultas:

-- Query historical data at a specific timestamp.
SELECT * FROM tt2 TIMESTAMP AS OF '2024-04-01 01:00:00';
-- Query historical data from 5 minutes before the current time.
SELECT * FROM tt2 TIMESTAMP AS OF CURRENT_TIMESTAMP() - 300;
-- Query historical data from the second-to-last commit.
SELECT * FROM tt2 TIMESTAMP AS OF GET_LATEST_TIMESTAMP('tt2', 2);

A figura a seguir ilustra o funcionamento de uma consulta time travel:

O exemplo utiliza uma tabela transacional chamada src:

• Lado esquerdo (processo de atualização de dados): As transações t1 a t5 geram, cada uma, um arquivo de dados delta. A COMPACTION é executada em t2 e t4, produzindo os arquivos compactados c1 e c2. Em c1, o registro histórico intermediário (2,a) é excluído e o registro mais recente (2,b) é retido.

• Resolução da consulta: Para consultar dados históricos em t1, o sistema lê apenas o arquivo delta d1. Para consultar em t2, ele lê o arquivo compactado c1 e retorna três registros. Para consultar em t3, ele lê c1 e o arquivo delta d3, mesclando-os para a saída. Execuções mais frequentes de COMPACTION aceleram as consultas, mas aumentam a sobrecarga operacional. Escolha uma política de acionamento adequada aos seus requisitos.

A sintaxe SQL suporta constantes, funções comuns e as cláusulas TIMESTAMP AS OF expr e VERSION AS OF expr para consultas históricas precisas. Para detalhes, consulte Consultas time travel.

Consultas incrementais

O MaxCompute projeta e desenvolve uma nova sintaxe de consulta incremental SQL para otimizar consultas incrementais e computação incremental em Delta tables. Após o envio de uma instrução de consulta incremental SQL, o engine do MaxCompute analisa as versões históricas de dados incrementais a serem consultadas, recupera os arquivos de dados compactados relevantes, mescla os dados dos arquivos e retorna a saída.

A figura a seguir mostra o processo de consulta incremental:

O exemplo usa a mesma tabela transacional src com as transações t1 a t5 e os arquivos compactados c1 (em t2) e c2 (em t4):

• Se begin for t1-1 e end for t1, o sistema lê apenas o arquivo delta d1 em t1.

• Se end for t2, o sistema lê os arquivos delta d1 e d2.

• Se begin for t1 e end for t2-1, o intervalo da consulta abrange de t1 a t2. Como não existem dados incrementais nesse intervalo, linhas vazias são retornadas.

Os dados nos arquivos compactados c1 e c2 gerados pela COMPACTION não são considerados novos dados para a saída da consulta incremental.

Para a sintaxe de consulta incremental e limites de parâmetros, consulte a seção "Parâmetros e limites de consultas incrementais" em Consultas time travel e consultas incrementais.

Data skipping otimizado baseado em chave primária

A distribuição de dados e os índices das Delta tables são construídos com base nos valores das colunas de chave primária. Ao consultar por chave primária, o sistema aplica filtros em vários níveis para reduzir significativamente a leitura de dados, melhorando a eficiência da consulta em centenas a milhares de vezes.

Exemplo: Uma Delta table contém 100 milhões de registros. Filtrar por um único valor de chave primária pode exigir a leitura de apenas 10.000 registros.

O processo de filtragem em três níveis:

1. Poda de buckets (Bucket pruning): Localiza o bucket que contém a chave primária alvo, eliminando varreduras em todos os outros buckets.

2. Poda de arquivos de dados: Dentro do bucket alvo, identifica apenas os arquivos de dados que contêm o valor da chave primária.

3. Filtragem de intervalo no nível de bloco: Aplica filtragem precisa com base na distribuição de valores da chave primária dentro dos blocos de arquivo, extraindo apenas os blocos que contêm o valor alvo.

Planos de consulta e análise SQL otimizados

Cada bucket em uma Delta table armazena dados únicos e ordenados pelo valor da chave primária. O Otimizador SQL aproveita essas propriedades para eliminar operações custosas:

Após eliminar DISTINCT, ordenação e shuffle global, o desempenho da consulta melhora em 100%.