O Flink SQL suporta joins complexos em tabelas dinâmicas com diversas semânticas de consulta e tipos de join. Evite criar um produto cartesiano, pois essa operação não tem suporte e causa falha nas consultas. O sistema não otimiza a ordem do join por padrão. Para obter melhor desempenho, liste as tabelas na cláusula FROM da menor para a maior frequência de atualização.
Visão geral dos joins
|
Tipo de join |
Descrição |
Restrições |
|
O regular join é o tipo mais versátil. Novos registros ou alterações em qualquer lado do join ficam visíveis e afetam todo o resultado. |
A sintaxe do regular join é flexível e aceita operações INSERT, UPDATE e DELETE nas tabelas de entrada. No entanto, esse join mantém os dados de entrada de ambos os lados em seu estado, que pode crescer indefinidamente. Para evitar tamanho excessivo de estado, defina um tempo de vida (TTL) do estado, mas isso pode comprometer a precisão dos resultados. |
|
|
O interval join retorna o produto cartesiano de registros que atendem tanto à condição de join quanto às restrições de tempo. |
Exige pelo menos uma condição de igualdade e uma condição de limite de tempo aplicável a ambos os lados. O intervalo de tempo pode ser definido como uma única condição (como <, <=, >= ou >), usando uma condição |
|
|
O temporal join associa um stream a uma tabela versionada. Ele usa event time ou processing time para correlacionar dados à versão correta da tabela em um ponto específico no tempo. |
Ambas as tabelas devem usar a mesma semântica de tempo (processing time ou event time). Atenção ao ciclo de vida do resultado do join: a condição geralmente se baseia em um timestamp específico. |
|
|
O lookup join normalmente enriquece uma tabela com dados consultados de um sistema externo. Esse join exige que uma tabela tenha um atributo de processing time e que a outra seja apoiada por uma tabela de dimensões. |
Uma tabela deve ter um atributo de processing time, enquanto a outra precisa usar um conector source de lookup. Também é necessária uma condição de igualdade entre as duas tabelas. |
|
|
O lateral join conecta uma tabela à saída de uma função com valor de tabela. Cada linha da tabela à esquerda é associada a todas as linhas produzidas pela chamada correspondente à função. |
Requer uma condição constante |
Regular joins
Os quatro tipos comuns de regular joins são:
INNER JOIN: Retorna registros que satisfazem a condição de join em ambas as tabelas (interseção).
LEFT JOIN: Retorna todos os registros da tabela à esquerda, mesmo sem correspondência na tabela à direita (preserva a tabela esquerda).
RIGHT JOIN: Retorna todos os registros da tabela à direita, mesmo sem correspondentes na tabela à esquerda (preserva a tabela direita).
FULL OUTER JOIN: Retorna a união das duas tabelas, incluindo registros correspondentes e não correspondentes.
Diagrama do regular join
Exemplo de regular join
Faça login no console do Realtime Compute for Apache Flink.
Na lista de workspaces, localize o workspace desejado e clique em Console na coluna Actions.
No painel de navegação à esquerda, clique em .
-
Clique no ícone
e, em seguida, clique em New Blank Stream Draft. Insira um name, selecione uma engine version e clique em Create.Este exemplo mostra como usar um join para correlacionar linhas entre tabelas, associando codinomes de super-heróis aos seus nomes reais.
CREATE TEMPORARY TABLE NOC ( agent_id STRING, codename STRING ) WITH ( 'connector' = 'faker', -- The Faker connector generates simulated data. 'fields.agent_id.expression' = '#{regexify ''(1|2|3|4|5){1}''}', -- Generates a random number from 1 to 5. 'fields.codename.expression' = '#{superhero.name}', -- A built-in Faker function that randomly generates a superhero name. 'number-of-rows' = '10' -- Specifies that 10 rows of data are generated. ); CREATE TEMPORARY TABLE RealNames ( agent_id STRING, name STRING ) WITH ( 'connector' = 'faker', 'fields.agent_id.expression' = '#{regexify ''(1|2|3|4|5){1}''}', 'fields.name.expression' = '#{Name.full_name}', -- A built-in Faker function that randomly generates a name. 'number-of-rows' = '10' ); SELECT name, codename FROM NOC INNER JOIN RealNames ON NOC.agent_id = RealNames.agent_id; -- If the agent_id (1-5) in both tables is equal, the name and codename are returned. -
No canto superior direito, clique em Debug, selecione um cluster de depuração e clique em OK. Caso não tenha um session cluster, consulte Create a session cluster.
Após a conclusão da depuração, o resultado será uma tabela com duas colunas, name e codename, exibindo os dados de saída do Regular Join.
Para mais informações sobre como usar regular joins, consulte Regular join statements.
Interval joins
Um interval join associa registros de dois streams que ocorrem dentro de um intervalo de tempo específico. Esse tipo de join geralmente correlaciona eventos de dois streams próximos temporalmente.
Diagrama do interval join
Exemplo de interval join
Este exemplo associa pedidos a envios. Ele filtra os resultados para incluir apenas pedidos enviados até três horas após a realização do pedido.
Crie um rascunho ETL conforme descrito em Regular joins.
CREATE TEMPORARY TABLE orders (
id INT,
order_time AS TIMESTAMPADD(HOUR, CAST(FLOOR(RAND()*(1-5+1)+5)*(-1) AS INT), CURRENT_TIMESTAMP) -- Randomly gets a time from 2, 3, or 4 hours ago based on the local time.
)
WITH (
'connector' = 'datagen', -- The datagen connector can periodically generate random data.
'rows-per-second'='10', -- The rate of random data generation, 10 rows/s.
'fields.id.kind'='sequence', -- The sequence generator.
'fields.id.start'='1', -- The sequence starts at 1.
'fields.id.end'='100' -- The sequence ends at 100.
);
CREATE TEMPORARY TABLE shipments (
order_id INT,
shipment_time AS TIMESTAMPADD(HOUR, CAST(FLOOR(RAND()*(1-5+1))+1 AS INT), CURRENT_TIMESTAMP) -- Randomly gets a time from 0, 1, or 2 hours ago based on the local time.
)
WITH (
'connector' = 'datagen',
'rows-per-second'='5',
'fields.order_id.kind'='sequence',
'fields.order_id.start'='1',
'fields.order_id.end'='100'
);
SELECT
o.id AS order_id,
o.order_time,
s.shipment_time,
TIMESTAMPDIFF(HOUR,o.order_time,s.shipment_time) AS hour_diff -- The time difference between order_time and shipment_time.
FROM orders o
JOIN shipments s ON o.id = s.order_id
WHERE
o.order_time BETWEEN s.shipment_time - INTERVAL '3' HOUR AND s.shipment_time; -- Filters for orders placed within three hours before the shipment time.No canto superior direito, clique em Debug e, em seguida, clique em OK. Os resultados da depuração são os seguintes.
Para mais informações sobre interval joins, consulte Interval join statements.
Temporal joins
O temporal join é utilizado com tabelas dinâmicas, cujo conteúdo muda ao longo do tempo. Ele associa um stream de eventos à versão da tabela dinâmica válida no momento de cada evento.
Diagrama do temporal join
Exemplo de temporal join
Este exemplo demonstra um cenário de negócios em que as taxas de câmbio variam em diferentes períodos e os pedidos devem ser calculados usando a taxa vigente no momento da transação.
Crie um novo rascunho ETL conforme descrito na seção Regular joins para ler os dados simulados durante a depuração.
CREATE TEMPORARY TABLE currency_rates (
`currency_code` STRING,
`eur_rate` DECIMAL(6,4),
`rate_time` TIMESTAMP(3),
WATERMARK FOR `rate_time` AS rate_time - INTERVAL '15' SECOND,
PRIMARY KEY (currency_code) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'topic' = 'currency_rates',
'properties.bootstrap.servers' = '${secret_values.kafkahost}',
'properties.auto.offset.reset' = 'earliest',
'properties.group.id' = 'currency_rates',
'key.format' = 'raw',
'value.format' = 'json'
);
CREATE TEMPORARY TABLE transactions (
`id` STRING,
`currency_code` STRING,
`total` DECIMAL(10,2),
`transaction_time` TIMESTAMP(3),
WATERMARK FOR `transaction_time` AS transaction_time - INTERVAL '30' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'transactions',
'properties.bootstrap.servers' = '${secret_values.kafkahost}',
'properties.auto.offset.reset' = 'earliest',
'properties.group.id' = 'transactions',
'key.format' = 'raw',
'key.fields' = 'id',
'value.format' = 'json'
);
SELECT
t.id,
t.total * c.eur_rate AS total_eur,
c.eur_rate,
t.total,
c.currency_code,
c.rate_time,
t.transaction_time
FROM transactions t
JOIN currency_rates FOR SYSTEM_TIME AS OF t.transaction_time AS c
ON t.currency_code = c.currency_code
;No canto superior direito, clique em Debug e, em seguida, clique em OK.
A taxa de câmbio mudou duas vezes, às 20:16:11 e às 20:35:22. Uma transação ocorreu às 20:35:14. Nesse momento, a taxa de câmbio ainda não havia sido atualizada. Portanto, o cálculo dessa transação deve usar a taxa de câmbio que entrou em vigor às 20:16:11.
Lookup joins
Um lookup join enriquece um stream de dados com dados estáticos ou de mudança lenta provenientes de um sistema externo. Por exemplo, é possível associar um stream de pedidos em tempo real a dados de referência, como informações de produtos armazenadas em um banco de dados relacional. Esse join exige que uma tabela tenha um atributo de processing time e que a outra esteja conectada por meio de um conector de lookup, como o conector MySQL.
Diagrama do lookup join
Inclua FOR SYSTEM_TIME AS OF PROCTIME() para associar cada registro do stream ao snapshot da tabela de dimensões naquele momento. Se os dados da tabela de dimensões mudarem posteriormente, os resultados do join não serão afetados.
A condição ON deve incluir uma condição de igualdade em um campo que a tabela de dimensões possa usar para lookups.
Exemplo de lookup join
Este exemplo mostra como enriquecer dados de pedidos com dados estáticos de um conector externo para adicionar nomes de produtos.
Crie um rascunho ETL conforme descrito em Regular joins.
CREATE TEMPORARY TABLE orders (
order_id STRING,
product_id INT,
order_total INT
) WITH (
'connector' = 'faker', -- The Faker connector generates simulated data.
'fields.order_id.expression' = '#{Internet.uuid}', -- Generates a random UUID.
'fields.product_id.expression' = '#{number.numberBetween ''1'',''5''}', -- Generates a random number from 1 to 5.
'fields.order_total.expression' = '#{number.numberBetween ''1000'',''5000''}', -- Generates a random number from 1000 to 5000.
'number-of-rows' = '10' -- The number of data rows to generate.
);
-- Connect to static product data in MySQL.
CREATE TEMPORARY TABLE products (
product_id INT,
product_name STRING
)
WITH(
'connector' = 'mysql',
'hostname' = '${secret_values.mysqlhost}',
'port' = '3306',
'username' = '${secret_values.username}',
'password' = '${secret_values.password}',
'database-name' = 'db2024',
'table-name' = 'products'
);
SELECT
o.order_id,
p.product_name,
o.order_total,
CASE
WHEN o.order_total > 3000 THEN 1
ELSE 0
END AS is_importance -- Adds the is_importance field. The value is 1 if the order total exceeds 3000, indicating an important order.
FROM orders o
JOIN products FOR SYSTEM_TIME AS OF PROCTIME() AS p -- The FOR SYSTEM_TIME AS OF PROCTIME() clause ensures that as the join operator processes rows from `orders`, each row from `orders` is joined with the `products` rows that match the join condition.
ON o.product_id = p.product_id;A tabela products contém 5 registros de exemplo: 1-cola, 2-orange, 3-milk, 4-tea, 5-coffee.
No canto superior direito, clique em Debug e, em seguida, clique em OK. Os resultados da depuração são os seguintes.
Para mais informações sobre lookup joins, consulte JOIN statements for dimension tables.
Lateral joins
Um lateral join utiliza uma subconsulta na cláusula FROM executada para cada linha da consulta externa. Isso pode melhorar a flexibilidade e o desempenho da consulta ao reduzir varreduras de tabelas. Contudo, essa operação pode causar degradação de desempenho se a consulta interna for complexa ou processar um grande volume de dados.
Exemplo de lateral join
Este exemplo agrega registros de vendas para identificar os três produtos com maior quantidade total vendida.
Crie um rascunho ETL conforme descrito em Regular joins.
CREATE TEMPORARY TABLE sale (
sale_id STRING,
product_id INT,
sale_num INT
)
WITH (
'connector' = 'faker', -- The Faker connector generates simulated data.
'fields.sale_id.expression' = '#{Internet.uuid}', -- Generates a random UUID.
'fields.product_id.expression' = '#{regexify ''(1|2|3|4|5){1}''}', -- Generates a random number from 1 to 5.
'fields.sale_num.expression' = '#{number.numberBetween ''1'',''10''}', -- Generates a random integer from 1 to 10.
'number-of-rows' = '50' -- Generates 50 rows of data.
);
CREATE TEMPORARY TABLE products (
product_id INT,
product_name STRING,
PRIMARY KEY(product_id) NOT ENFORCED
)
WITH(
'connector' = 'mysql',
'hostname' = '${secret_values.mysqlhost}',
'port' = '3306',
'username' = '${secret_values.username}',
'password' = '${secret_values.password}',
'database-name' = 'db2024',
'table-name' = 'products'
);
SELECT
p.product_name,
s.total_sales
FROM products p
LEFT JOIN LATERAL
(SELECT SUM(sale_num) AS total_sales FROM sale WHERE sale.product_id = p.product_id) s ON TRUE
ORDER BY total_sales DESC
LIMIT 3;No canto superior direito, clique em Debug e, em seguida, clique em OK. Os resultados da depuração são os seguintes.
Documentos relacionados
Se você considera apenas o processing time dos eventos, e não o event time, use um temporal join baseado em processing time. Para mais informações, consulte Processing-time temporal join statements.
Para mais detalhes sobre como usar regular joins, consulte Regular join statements.
Saiba mais sobre o uso de interval joins em Interval join statements.
Para informações adicionais sobre lookup joins, consulte JOIN statements for dimension tables.