Quando vários streams do Flink atualizam colunas diferentes na mesma linha, uma operação UPSERT padrão substitui a linha inteira. Assim, o stream posterior apaga silenciosamente as alterações feitas pelo stream anterior. As atualizações parciais de colunas permitem que cada stream grave apenas as colunas sob sua responsabilidade. Dessa forma, as atualizações simultâneas de streams independentes são preservadas na mesma linha sem conflitos.
Quando usar este recurso
Streams independentes, mesma linha: dois streams atualizam colunas diferentes no mesmo registro simultaneamente. Sem atualizações parciais de colunas, a gravação posterior substitui as colunas definidas pela gravação anterior, causando perda de dados.
Gravações esparsas: um stream fornece valores apenas para um subconjunto de colunas. Sem atualizações parciais de colunas, as colunas não especificadas são definidas como NULL, destruindo os dados existentes.
Como funciona
O UPSERT combina INSERT e UPDATE em uma única operação. Todo registro processado por UPSERT deve incluir as colunas de chave primária.
Se não existir nenhum registro com a chave primária fornecida, o UPSERT insere um novo registro.
Caso já exista um registro com a chave primária informada, o UPSERT o atualiza com os novos valores.
No processamento de streams com junções de várias tabelas, diferentes streams podem atualizar colunas distintas na mesma linha. O UPSERT padrão substitui a linha inteira, portanto, as atualizações de um stream podem sobrescrever as alterações de outro. As atualizações parciais de colunas resolvem esse problema ao restringir a gravação de cada stream apenas às colunas que ele gerencia.
Modos de atualização parcial de colunas
O MaxCompute oferece dois modos de atualização parcial de colunas para o Flink Connector.
|
Modo |
Como funciona |
Quando usar |
|
Modo dinâmico |
Detecta automaticamente as colunas não NULL em cada registro e atualiza apenas essas colunas. As colunas NULL permanecem inalteradas. |
O conjunto de colunas é desconhecido antecipadamente ou varia por registro |
|
Modo estático |
Atualiza somente as colunas que você especificar explicitamente. Todas as outras colunas mantêm seus valores existentes. |
O conjunto de colunas é fixo e conhecido antecipadamente |
A tabela a seguir mostra como a mesma sequência de gravações produz resultados diferentes em cada modo. A coluna a é a chave primária.
|
Modo |
Dados iniciais |
Após gravar (a, b, c) |
Após gravar (a, d, NULL) |
Após gravar (a, NULL, e) |
|
Modo regular |
(null, null, null) |
(a, b, c) |
(a, d, null) |
(a, null, e) |
|
Modo dinâmico |
(null, null, null) |
(a, b, c) |
(a, d, c) |
(a, d, e) |
|
Modo estático (segunda coluna especificada) |
(null, null, null) |
(a, b, null) |
(a, d, null) |
(a, null, null) |
Pré-requisitos
Antes de começar, verifique se você tem:
Um projeto do MaxCompute com uma tabela Delta que tenha atualizações parciais de colunas ativadas
Flink Connector configurado para se conectar ao MaxCompute
Ative as atualizações parciais de colunas em uma tabela Delta
Defina acid.partial.fields.update.enable=true em TBLPROPERTIES ao criar a tabela Delta. Para obter a lista completa de parâmetros de tabelas Delta, consulte Parâmetros para tabelas Delta.
CREATE TABLE IF NOT EXISTS partial_upsert_test
(pk INT NOT NULL,
c1 STRING,
c2 STRING,
c3 STRING,
PRIMARY KEY(pk))
TBLPROPERTIES('transactional'='true', 'acid.partial.fields.update.enable'='true');Configure o Flink Connector para atualizações parciais de colunas
Parâmetros
|
Parâmetro |
Descrição |
|
|
Ativa as atualizações parciais de colunas. Quando |
|
|
Especifique as colunas a serem atualizadas (modo estático). As colunas de chave primária são sempre incluídas automaticamente. |
O parâmetro upsert.partial-column.name deve usar os nomes das colunas da tabela do MaxCompute, e não os nomes das colunas da tabela interna do Flink.
Não é possível adicionar nomes de colunas de chave de partição a upsert.partial-column.name .
Exemplo de modo dinâmico
A definição de tabela Flink a seguir ativa as atualizações parciais de colunas dinâmicas. O sistema detecta automaticamente quais colunas contêm valores não NULL e atualiza apenas essas colunas.
CREATE TABLE partialtable (
pk INT,
c1 STRING,
c2 STRING,
c3 STRING,
PRIMARY KEY(pk) NOT ENFORCED
) WITH (
'connector' = 'maxcompute',
'odps.end.point' = 'https://service.cn-hangzhou-vpc.maxcompute.aliyun-inc.com/api', -- VPC endpoint
'odps.project.name' = 'project_name',
'odps.namespace.schema' = 'true', -- Enable the three-layer model
'table.name' = 'project.schema.tablename',
'sink.operation' = 'upsert',
'upsert.write.bucket.num' = '1',
'upsert.partial-column.enable' = 'true', -- Enable dynamic mode
'odps.access.id' = '<your-access-key-id>',
'odps.access.key' = '<your-access-key-secret>'
);O exemplo a seguir demonstra como o modo dinâmico preserva os valores existentes em três gravações sequenciais.
Etapa 1: insira o registro inicial. Dados após esta etapa: [1, a, b, c].
INSERT INTO partialtable VALUES (1, 'a', 'b', 'c');Etapa 2: atualize apenas c2. Como c1 e c3 não foram especificados, eles mantêm seus valores. Dados após esta etapa: [1, a, d, c].
INSERT INTO partialtable(pk, c2) VALUES (1, 'd');Etapa 3: atualize apenas c3. Como c1 e c2 não foram especificados, eles mantêm seus valores. Dados após esta etapa: [1, a, d, e].
INSERT INTO partialtable(pk, c3) VALUES (1, 'e');Exemplo de modo estático
A definição de tabela Flink a seguir restringe as atualizações apenas à coluna c2. Todas as gravações nesta tabela afetam c2, independentemente de quais outros valores forem fornecidos.
CREATE TABLE PartialTable2 (
pk INT,
c1 STRING,
c2 STRING,
c3 STRING,
PRIMARY KEY(pk) NOT ENFORCED
) WITH (
'connector' = 'maxcompute',
'odps.end.point' = 'https://service.cn-hangzhou-vpc.maxcompute.aliyun-inc.com/api', -- VPC endpoint
'odps.project.name' = 'project_name',
'odps.namespace.schema' = 'true', -- Enable the three-layer model
'table.name' = 'project.schema.tablename',
'sink.operation' = 'upsert',
'upsert.write.bucket.num' = '1',
'upsert.partial-column.enable' = 'true',
'upsert.partial-column.name' = 'c2', -- Only update c2; primary key is included by default
'odps.access.id' = '<your-access-key-id>',
'odps.access.key' = '<your-access-key-secret>'
);Próximos passos
Para saber mais sobre como gravar dados em tabelas Delta com o Flink, consulte Usar o Flink para gravar dados em uma tabela Delta.
Para obter uma referência completa sobre o recurso de atualização parcial de colunas em tabelas Delta, consulte Atualizar dados em colunas específicas.