Hologres：分布鍵Distribution Key

在Hologres中，Distribution Key屬性指定了表資料的分布策略，系統會保證Distribution Key相同的記錄被分配到同一個Shard上。建表時設定文法如下：

-- Hologres V2.1版本起支援的文法
CREATE TABLE <table_name> (...) WITH (distribution_key = '[<columnName>[,...]]');

-- 所有版本支援的文法
BEGIN;
CREATE TABLE <table_name> (...);
call set_table_property('<table_name>', 'distribution_key', '[<columnName>[,...]]');
COMMIT;

參數說明如下：

Distribution Key是非常重要的分布式概念，合理設定Distribution Key可以達到如下效果：

• 顯著提高計算效能。

  不同的Shard可以進行並行計算，從而提高計算效能。

• 顯著提高每秒查詢率（QPS）。

  當您以Distribution Key做過濾條件時，Hologres可以直接篩選出資料相關的Shard進行掃描。否則Hologres需要讓所有的Shard參與計算，影響QPS。

• 顯著提高Join效能。

  當兩張表在同一個Table Group內，並且Join的欄位是Distribution Key時，那麼資料分布保證表A一個Shard內的資料和表B同一Shard內的資料對應，只需要直接在本節點Join本節點資料（Local Join）即可，可以顯著提高執行效率。

Distribution Key設定原則總結如下：

• Distribution Key盡量選擇分布均勻的欄位，否則容易因為資料扭曲導致負載傾斜，使得查詢效率變低，排查資料扭曲請參見查看Worker傾斜關係。

• 選擇Group By頻繁的欄位作為Distribution Key。

• Join情境中，設定Join欄位為Distribution Key，實現Local Join，避免資料Shuffle。同時進行Join的表需要在同一個Table Group內。

• 不建議為一個表設定多個Distribution Key，建議設定的Distribution Key不超過兩個欄位。設定多欄位為Distribution Key，查詢時若沒有全部命中，容易出現資料Shuffle。

• 支援單列或者多列設定為Distribution Key，指定列時如設定單列，命令語句中不要保留多餘空格；如設定多個列，則以半形逗號（,）分隔，同樣不要保留多餘空格。指定多列為Distribution Key時，列的順序不影響資料的布局和查詢效能。

• 表設定了主鍵（PK）時，Distribution Key必須為PK或者PK中的部分欄位（不可為空，即不指定任何列），因為要求同一記錄的資料只能屬於一個Shard。如果沒有額外指定Distribution Key，預設將PK設定為Distribution Key。

為表設定了Distribution Key之後，資料會根據Distribution Key被分配到各個Shard上，演算法為Hash(distribution_key)%shard_count，結果為對應的Shard。系統會保證Distribution Key相同的記錄會被分配到同一個Shard上，樣本如下。

• V2.1版本起支援的建表文法：

  --設定a列為distribution key，系統會對a列的值做hash操作，再模數，即hash(a)%shard_count = shard_id，結果相同的一組資料分布在同一個Shard內
  CREATE TABLE tbl (
      a int NOT NULL,
      b text NOT NULL
  )
  WITH (
      distribution_key = 'a'
  );
  
  --設定a、b兩列為distribution key，系統對a,b兩個列的值做hash操作，再模數，即hash(a,b)%shard_count = shard_id，結果相同的一組資料分布在同一個Shard內
  CREATE TABLE tbl (
      a int NOT NULL,
      b text NOT NULL
  )
  WITH (
      distribution_key = 'a,b'
  );

• 所有版本支援的建表文法：

  --設定a列為distribution key，系統會對a列的值做hash操作，再模數，即hash(a)%shard_count = shard_id，結果相同的一組資料分布在同一個Shard內
  begin;
  create table tbl (
  a int not null,
  b text not null
  );
  call set_table_property('tbl', 'distribution_key', 'a');
  commit;
  
  --設定a、b兩列為distribution key，系統對a,b兩個列的值做hash操作，再模數，即hash(a,b)%shard_count = shard_id，結果相同的一組資料分布在同一個Shard內
  begin;
  create table tbl (
    a int not null,
    b text not null
  );
  call set_table_property('tbl', 'distribution_key', 'a,b');
  commit;
                              

資料分布示意圖如下：但在設定Distribution Key時，需要關注設定為Distribution Key欄位的資料最好是分布均勻的。Hologres的Shard數和Worker節點數有一定的關聯關係，詳情請參見基本概念。如果設定了資料分布不均勻的欄位作為Distribution Key之後，那麼資料會集中分布在某些Shard上，導致大部分的計算集中到部分Worker上，出現長尾效應，查詢效率降低。排查以及處理資料的傾斜情況詳情請參見查看Worker傾斜關係。

不設定Distribution Key時，資料將會被隨機分布在各個Shard，相同的資料可能會在相同Shard，也可能在不同的Shard，樣本如下。

--不設定distribution key
begin;
create table tbl (
a int not null,
b text not null
);
commit;

資料分布示意圖如下：

為表設定了Distribution Key，那麼相同的資料就分布在相同的Shard上，同時對於Group By彙總情境，資料在計算時按照設定的Distribution Key重新分配，因此可以將Group By頻繁的欄位設定為Distribution Key，這樣資料在Shard內就已經彙總，減少資料在Shard間的重分配，提高查詢效能，樣本如下。

• V2.1版本起支援的建表文法：

  CREATE TABLE agg_tbl (
      a int NOT NULL,
      b int NOT NULL
  )
  WITH (
      distribution_key = 'a'
  );
  
  --樣本查詢，對a列做彙總查詢
  select a,sum(b) from agg_tbl group by a;

• 所有版本支援的建表文法：

  begin;
  create table agg_tbl (
  a int not null,
  b int not null
  );
  call set_table_property('agg_tbl', 'distribution_key', 'a');
  commit;
  
  --樣本查詢，對a列做彙總查詢
  select a,sum(b) from agg_tbl group by a;

通過查看執行計畫（explain SQL），如下所示執行計畫結果中沒有redistribution運算元，說明資料沒有重分布。

多表關聯的情境比較複雜，可以遵循如下原則：

• 每個表的Join欄位都相同，那麼將Join欄位都設定為Distribution Key。

• 每個表的Join欄位不同，優先考慮大表間的Join，將大表的Join欄位設定為Distribution Key。

通過以下幾種情況舉例說明（本文中以三個表Join為例說明，大於三個表的Join情況可以參考本樣本）。

• 三個表的Join欄位相同

  在三個表Join的情境中，三個表的Join欄位都相同，那麼這種情況是最簡單的，可以直接將三個表的Join欄位都設定為Distribution Key，實現Local Join的能力。

  • V2.1版本起支援的建表文法：

    BEGIN;
    CREATE TABLE join_tbl1 (
        a int NOT NULL,
        b text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    CREATE TABLE join_tbl2 (
        a int NOT NULL,
        d text NOT NULL,
        e text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    CREATE TABLE join_tbl3 (
        a int NOT NULL,
        e text NOT NULL,
        f text NOT NULL,
        g text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    COMMIT;
    
    --3表join查詢
    SELECT * FROM join_tbl1
    INNER JOIN join_tbl2 ON join_tbl2.a = join_tbl1.a
    INNER JOIN join_tbl3 ON join_tbl2.a = join_tbl3.a;

  • 所有版本支援的建表文法：

    begin;
    create table join_tbl1(
    a int not null,
    b text not null
    );
    call set_table_property('join_tbl1', 'distribution_key', 'a');
    
    create table join_tbl2(
    a int not null,
    d text not null,
    e text not null
    );
    call set_table_property('join_tbl2', 'distribution_key', 'a');
    
    create table join_tbl3(
    a int not null,
    e text not null,
    f text not null,
    g text not null
    );
    call set_table_property('join_tbl3', 'distribution_key', 'a');
    commit;
    
    --3表join查詢
    SELECT * FROM join_tbl1
    INNER JOIN join_tbl2 ON join_tbl2.a = join_tbl1.a
    INNER JOIN join_tbl3 ON join_tbl2.a = join_tbl3.a;

  通過查看執行計畫（explain SQL），如下所示執行計畫中：

  • 沒有redistribution運算元，說明資料沒有重分布，實現了Local Join。

  • exchange運算元代表檔案層級彙總到Shard層級彙總，這樣就只需要對應Shard的資料，提升資料的查詢效率。

  

• 三個表的Join欄位不同

  在實際業務中，多表關聯時會有Join欄位不相同的情境，這個時候可以根據如下原則來設定Distribution Key：

  • 核心最佳化原則是優先考慮大表間的Join，設定大表的Join欄位為Distribution Key；小表因其資料量較少，無需過多考慮。

  • 表資料量大致相同的情況，可以設定Group By頻繁的Join欄位為Distribution Key。

  如下樣本，有三個表相互Join，Join的欄位不完全一樣，這個時候選擇大表的Join欄位為Distribution Key，join_tbl_1這個表資料量有一千萬條，join_tbl_2和join_tbl_3分別有一百萬條，以join_tbl_1為主要最佳化對象。

  • V2.1版本起支援的建表文法：

    BEGIN;
    -- join_tbl_1為1kw資料量
    CREATE TABLE join_tbl_1 (
        a int NOT NULL,
        b text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    -- join_tbl_2為100w資料量
    CREATE TABLE join_tbl_2 (
        a int NOT NULL,
        d text NOT NULL,
        e text NOT NULL
    )
    WITH (
        distribution_key = 'a'
    );
    -- join_tbl_3為100w資料量
    CREATE TABLE join_tbl_3 (
        a int NOT NULL,
        e text NOT NULL,
        f text NOT NULL,
        g text NOT NULL
    );
     WITH (
         distribution_key = 'a'
     );
    COMMIT;
    
    --join key不相同時，選擇大表的join key為distribution key。
    SELECT * FROM join_tbl_1
    INNER JOIN join_tbl_2 ON join_tbl_2.a = join_tbl_1.a
    INNER JOIN join_tbl_3 ON join_tbl_2.d = join_tbl_3.f;

  • 所有版本支援的建表文法：

    begin;
    --join_tbl_1為1kw資料量
    create table join_tbl_1(
    a int not null,
    b text not null
    );
    call set_table_property('join_tbl_1', 'distribution_key', 'a');
    
    --join_tbl_2為100w資料量
    create table join_tbl_2(
    a int not null,
    d text not null,
    e text not null
    );
    call set_table_property('join_tbl_2', 'distribution_key', 'a');
    
    --join_tbl_3為100w資料量
    create table join_tbl_3(
    a int not null,
    e text not null,
    f text not null,
    g text not null
    );
    --call set_table_property('join_tbl_3', 'distribution_key', 'a');
    commit;
    
    --join key不相同時，選擇大表的join key為distribution key。
    SELECT * FROM join_tbl_1
    INNER JOIN join_tbl_2 ON join_tbl_2.a = join_tbl_1.a
    INNER JOIN join_tbl_3 ON join_tbl_2.d = join_tbl_3.f;

  通過查看執行計畫（explain SQL），如下所示執行計畫表明：

  • 在join_tbl_2和join_tbl_3表之間有redistribution運算元，因為join_tbl_3是小表，Join欄位與Distribution Key不一致，所以資料進行了重分布。

  • join_tbl_1和join_tbl_2表之間沒有redistribution運算元，因為兩表將Join欄位都設定為Distribution Key，因此資料不會重分布。