維表JOIN語句

背景資訊

大部分連接器的維表Join都可以使用Cache策略，不同連接器對Cache策略的支援情況稍有不同，請查看對應的連接器文檔確定具體的支援情況。通用的Cache策略詳情如下：

• None（預設值）：無緩衝。

• LRU：緩衝維表裡的部分資料。源表的每條資料都會觸發系統先在Cache中尋找資料，如果沒有找到，則去物理維表中尋找。

• ALL：緩衝維表裡的所有資料。在Job運行前，系統會將維表中所有資料載入到Cache中，之後所有的維表尋找資料都會通過Cache進行。如果在Cache中無法找到資料，則KEY不存在。全量的Cache有一個到期時間，到期後會重新載入一遍全量Cache。適用於遠端資料表資料量小且MISS KEY（源表資料和維表JOIN時，ON條件無法關聯）特別多的情境。

使用限制

• 維表JOIN僅支援對當前時刻維錶快照的關聯。

• 維表支援INNER JOIN和LEFT JOIN，不支援RIGHT JOIN或FULL JOIN。

注意事項

• 如果您有一對一JOIN需求，請確保串連條件中包含了維表中具有唯一性欄位的等值串連條件。

• 對每條流式資料，只會關聯當時維表的最新版本資料，即JOIN行為只發生在處理時間（Processing Time）。如果JOIN行為發生後，維表中的資料發生了變化（新增、更新或刪除），則已關聯的維表資料不會被同步變化。具體的維表的行為請參見對應連接器行為。

維表JOIN文法

SELECT column-names
FROM table1 [AS <alias1>]
[LEFT] JOIN table2 FOR SYSTEM_TIME AS OF PROCTIME() [AS <alias2>]
ON table1.column-name1 = table2.key-name1;

使用樣本

• 測試資料

  • 表1 kafka_input

    id（bigint）	name（varchar）	age（bigint）

    id（bigint）	name（varchar）	age（bigint）
    1	lilei	22
    2	hanmeimei	20
    3	libai	28

  • 表2 phoneNumber

    name（varchar）	phoneNumber（bigint）

    name（varchar）	phoneNumber（bigint）
    dufu	1390000111
    baijuyi	1390000222
    libai	1390000333
    lilei	1390000444

• 測試語句

  CREATE TEMPORARY TABLE kafka_input (
    id   BIGINT,
    name VARCHAR,
    age  BIGINT
  ) WITH (
    'connector' = 'kafka',
    'topic' = '<yourTopic>',
    'properties.bootstrap.servers' = '<yourKafkaBrokers>',
    'properties.group.id' = '<yourKafkaConsumerGroupId>',
    'format' = 'csv'
  );
  
  CREATE TEMPORARY TABLE phoneNumber(
    name VARCHAR,
    phoneNumber BIGINT,
    PRIMARY KEY(name) NOT ENFORCED
  ) WITH (
    'connector' = 'mysql',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  CREATE TEMPORARY TABLE result_infor(
    id BIGINT,
    phoneNumber BIGINT,
    name VARCHAR
  ) WITH (
    'connector' = 'blackhole'
  );
  
  INSERT INTO result_infor
  SELECT
    t.id,
    w.phoneNumber,
    t.name
  FROM kafka_input as t
  JOIN phoneNumber FOR SYSTEM_TIME AS OF PROCTIME() as w
  ON t.name = w.name;

• 測試結果

  id（bigint）	phoneNumber（bigint）	name（varchar）

  id（bigint）	phoneNumber（bigint）	name（varchar）
  1	1390000444	lilei
  3	1390000333	libai