Realtime Compute for Apache Flink:Criar e usar tabelas materializadas

CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
-- Primary key constraint
[([CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED)]

[COMMENT table_comment]
-- Partition key
[PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
-- With options
[WITH (key1=val1, key2=val2, ...)]
-- Data freshness
FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY }
-- Refresh mode
[REFRESH_MODE = { CONTINUOUS | FULL }]
AS  <select_statement>

Limitações

• Modifique a consulta apenas para tabelas materializadas criadas no VVR 11.1 ou posterior.

• Ao modificar uma consulta, adicione apenas colunas e modifique a lógica de computação. Não altere a ordem das colunas existentes nem modifique suas definições.

  Operação	Suportado	Descrição
  Adicionar uma nova coluna	Sim	Anexe novas colunas ao schema mantendo a ordem existente das colunas.
  Modificar a lógica de computação de uma coluna existente (sem alterar seu nome ou tipo)	Sim	Modifique a lógica de computação, mas mantenha o nome da coluna e o tipo de dados inalterados.
  Alterar a ordem das colunas existentes	Não	A ordem das colunas é fixa. Para alterá-la, exclua e recrie a tabela materializada.
  Modificar o nome ou o tipo de dados de uma coluna existente	Não	Exclua e recrie a tabela materializada.

Exemplo

1. Clique em Edit Table e modifique a consulta. O código a seguir fornece um exemplo:

   ALTER MATERIALIZED TABLE `paimon`.`default`.`mt-orders`
       AS
       SELECT
         *,
         price * quantity AS total_price
       FROM orders
       WHERE price * quantity > 1000
   ;

2. Clique em Preview para ver uma comparação antes e depois.

   

3. Clique em OK. Visualize a coluna recém-adicionada e a lógica de consulta modificada na aba Table Schema.

Atualizações incrementais

CREATE MATERIALIZED TABLE mt_shipped_orders (
    PRIMARY KEY (order_id) NOT ENFORCED
)
FRESHNESS = INTERVAL '30' MINUTE
AS
SELECT 
    order_id,
    COALESCE(customer_id, 'Unknown') AS customer_id,
    CAST(order_amount AS DECIMAL(10, 2)) AS order_amount,
    CASE 
        WHEN status = 'shipped' THEN 'Completed'
        WHEN status = 'pending' THEN 'In Progress'
        ELSE 'Unknown'
    END AS order_status,
    DATE_FORMAT(order_ts, 'yyyyMMdd') AS order_date,
    UDSF_ProcessFunction(notes) AS notes
FROM 
    orders
WHERE
    status = 'shipped';

CREATE MATERIALIZED TABLE mt_enriched_orders (
    PRIMARY KEY (order_id, order_tag) NOT ENFORCED
)
FRESHNESS = INTERVAL '30' MINUTE
AS
WITH o AS (
    SELECT
        order_id,
        product_id,
        quantity,
        proc_time,
        e.tag AS order_tag
    FROM 
        orders,
        LATERAL TABLE(UDTF_StringSplitFunction(tags, ',')) AS e(tag))
SELECT 
    o.order_id,
    o.product_id,
    p.product_name,
    p.category,
    o.quantity,
    p.price,
    o.quantity * p.price AS total_amount,
    order_tag
FROM o 
LEFT JOIN 
    product_info FOR SYSTEM_TIME AS OF PROCTIME() AS p
ON 
    o.product_id = p.product_id;

Backfill de dados

Anteriormente, corrigir resultados de processamento de fluxo para dados históricos exigia o desenvolvimento de um job em lote separado. Com tabelas materializadas, faça diretamente o backfill de partições de dados históricos. Essa abordagem unifica o processamento em lote e em streaming, reduzindo custos de desenvolvimento e O&M.

1. Clique na tabela materializada desejada dentro de seu catálogo.

2. Na aba Data Information, faça o backfill dos dados.

   Se você definiu uma coluna de particionamento ao criar a tabela materializada, ela será uma tabela particionada. Caso contrário, será uma tabela não particionada.

   Tabela particionada

   Na seção Data Partitions, clique em Trigger Update se esta for a primeira vez que você executa o backfill de dados ou se a partição necessária não existir. Se as partições já existirem, selecione uma partição específica e clique em Refresh na coluna Actions.

   

   

   Parâmetros

   • Coluna de particionamento: A coluna de particionamento da tabela. Por exemplo, se você inserir 20241201, todos os dados na partição ds=20241201 passarão por backfill.

   • Nome da Tarefa: O nome da tarefa de backfill de dados.

   • Intervalo de Atualização (Opcional): Especifica se as atualizações devem ser propagadas em cascata para tabelas materializadas downstream. A partir da tabela atual, todas as tabelas materializadas na linhagem de dados são atualizadas. A profundidade máxima suportada da linhagem downstream é de 6 níveis.

     Nota

     • Ao atualizar uma tabela particionada, as tabelas materializadas downstream devem ter exatamente as mesmas colunas de particionamento da tabela inicial. Caso contrário, a operação de atualização falhará.

     • Se uma atualização falhar para qualquer tabela materializada na linhagem, todos os nós downstream subsequentes também falharão.

   • Destino da Implantação: Selecione uma fila ou um cluster de sessão. A seleção padrão é default-queue.

   Tabela não particionada

   Na seção Data Status, clique em Refresh.

   

   Parâmetros

   • Nome da Tarefa: O nome da tarefa de backfill de dados.

   • Intervalo de Atualização: Esta opção não está disponível para tabelas não particionadas.

     Nota

     • Durante uma atualização, os dados downstream são totalmente atualizados.

     • Se uma atualização falhar para qualquer tabela materializada na linhagem, todos os nós downstream subsequentes também falharão.

     • Atualizações em cascata não são suportadas se a tabela inicial for uma tabela não particionada atualizada por um job de streaming.

   • Destino da Implantação: Selecione uma fila ou um cluster de sessão. A seleção padrão é default-queue.

3. Backfill agendado e em lote.

   Use a Orquestração de Tarefas para criar um workflow para a tabela materializada executar jobs de backfill conforme um agendamento. Utilize também o recurso de backfill de dados do workflow para fazer backfill de dados em massa para um intervalo de tempo especificado.

Modificar a atualização dos dados

1. Sob o catálogo correspondente, clique no banco de dados da materialized table e, em seguida, clique na materialized table desejada.

2. No canto superior direito, clique em Edit Freshness.

   • Se a tabela materializada não tiver uma chave primária, não será possível alternar seu método de atualização entre streaming e lote. O sistema usa um job de streaming para valores de atualização de dados inferiores a 30 minutos e um job em lote para valores de 30 minutos ou mais. Portanto, alterar a atualização além desse limite de 30 minutos não é permitido para tabelas sem chave primária.

   • Se a tabela upstream for uma tabela materializada, garanta que a atualização dos dados da tabela downstream seja um múltiplo inteiro positivo da atualização da tabela upstream.

   • A atualização máxima dos dados é de um dia.

Visualizar linhagem de dados

No painel de navegação à esquerda, escolha Operations Center > Data Lineage para acessar a página de linhagem de dados das tabelas materializadas. Nesta página, visualize as relações de linhagem entre todas as tabelas materializadas. Execute operações como Start/Stop Update e Edit Freshness diretamente em uma tabela materializada. Clique em Details para navegar até a página de detalhes da tabela materializada correspondente.