MaxCompute:データスキューのチューニング

次のコードは、GROUP BY 句を使用した擬似コードの例です。

SELECT  shop_id
        ,sum(is_open) AS open_days
FROM    table_xxx_di
WHERE   dt BETWEEN '${bizdate_365}' AND '${bizdate}'
GROUP BY shop_id;

データスキューが発生した場合、次の 3 つの解決策のいずれかを使用できます。

• 方法 1：GROUP BY のスキュー対策パラメーターを設定する。

  SET odps.sql.groupby.skewindata=true;

• 方法 2：乱数を追加する。

  この解決策は、SQL を書き換えて乱数を追加し、ロングテールを引き起こすキーを分割します。これは、GROUP BY 操作におけるロングテールを解決するための効果的な方法です。

  SQL クエリ Select Key,Count(*) As Cnt From TableName Group By Key; の場合、コンバイナがないと、mapper ノードはデータを reducer ノードにシャッフルし、reducer ノードが COUNT 操作を実行します。対応する実行計画は M->R です。

  ロングテールキーが特定されたと仮定すると、そのキーの作業を次のように再分散できます。

  -- ロングテールキーが KEY001 であると仮定します。
  SELECT  a.Key
          ,SUM(a.Cnt) AS Cnt
  FROM(SELECT  Key
              ,COUNT(*) AS Cnt
              FROM    <TableName>
              GROUP BY Key
              ,CASE WHEN KEY = 'KEY001' THEN Hash(Random()) % 50
               ELSE 0
              END
          ) a
  GROUP BY a.Key;

  変更された実行計画は M->R->R になります。実行ステップ数は増えますが、ロングテールキーが 2 段階で処理されるため、全体のランタイムは短縮される可能性があります。リソース消費と時間効率は方法 1 と同様です。ただし、実際のシナリオでは、ロングテールキーが複数存在することがよくあります。ロングテールキーを見つけて SQL を書き換える手間を考えると、方法 1 の方がコスト効率が高い場合が多いです。

• 方法 3：ローリングテーブルを作成する。

  コストを削減し、効率を向上させるために、過去 1 年間のデータを取得する必要がある場合があります。オンラインタスクの場合、毎回 T-1 から T-365 までのすべてのパーティションを読み取ることは、リソースの大きな無駄です。ローリングテーブルを作成すると、過去 1 年間のデータ取得に影響を与えることなく、読み取るパーティションの数を減らすことができます。次のコードに例を示します。

  まず、365 日分のマーチャントビジネスデータを GROUP BY 集約で初期化し、データ更新日をマークして、テーブル a として保存します。その後のオンラインタスクでは、T-2 日のテーブル a を table_xxx_di テーブルと結合し、再度 GROUP BY を実行できます。これにより、毎日読み取るパーティションの数が 365 から 2 に減少します。プライマリキー shop_id の重複が大幅に減少し、リソース消費も削減されます。

  -- ローリングテーブルを作成します。
  CREATE TABLE IF NOT EXISTS m_xxx_365_df
  (
    shop_id STRING,
    last_update_ds STRING,
    `365d_open_days` BIGINT
  )
  PARTITIONED BY
  (
    ds STRING COMMENT 'Date partition'
  )LIFECYCLE 7;
  -- 365 日の期間が 2021-05-01 から 2022-05-01 であると仮定します。1 回限りの初期化を実行します。
  INSERT OVERWRITE TABLE m_xxx_365_df PARTITION(ds = '20220501')
    SELECT shop_id,
           max(ds) as last_update_ds,
           sum(is_open) AS `365d_open_days`
    FROM table_xxx_di
    WHERE dt BETWEEN '20210501' AND '20220501'
    GROUP BY shop_id;
  -- 次に、実行される日次オンラインタスクは次のとおりです。
  INSERT OVERWRITE TABLE m_xxx_365_df PARTITION(ds = '${bizdate}')
    SELECT aa.shop_id, 
           aa.last_update_ds, 
           `365d_open_days` - COALESCE(is_open, 0) AS `365d_open_days` -- 開店日数の無限ローリングを防ぎます。
    FROM (
      SELECT shop_id, 
             max(last_update_ds) AS last_update_ds, 
             sum(`365d_open_days`) AS `365d_open_days`
      FROM (
        SELECT shop_id,
               ds AS last_update_ds,
               sum(is_open) AS `365d_open_days`
        FROM table_xxx_di
        WHERE ds = '${bizdate}'
        GROUP BY shop_id
        UNION ALL
        SELECT shop_id,
               last_update_ds,
               `365d_open_days`
        FROM m_xxx_365_df
        WHERE dt = '${bizdate_2}' AND last_update_ds >= '${bizdate_365}'
        -- ソースがすでにグループ化されている場合、ここでは GROUP BY は不要です。
      )
      GROUP BY shop_id
    ) AS aa
    LEFT JOIN (
      SELECT shop_id,
             is_open
      FROM table_xxx_di
      WHERE ds = '${bizdate_366}'
    ) AS bb
    ON aa.shop_id = bb.shop_id;
                              

あるテーブルに次のようなデータ分布があるとします。

次のステートメントを使用すると、データスキューが容易に発生する可能性があります。

SELECT  ds
        ,COUNT(DISTINCT shop_id) AS cnt
FROM    demo_data0
GROUP BY ds;

解決策は次のとおりです。

• 方法 1：パラメーターチューニング。

  次のパラメーターを設定します。

  SET odps.sql.groupby.skewindata=true;

• 方法 2：汎用的な 2 段階集計。

  shop_id フィールドのデータが不均等に分散している場合、方法 1 は効果的ではありません。より汎用的な方法は、パーティションフィールドの値に乱数を追加することです。

  -- 方法 A：乱数を連結します。CONCAT(ROUND(RAND(),1)*10,'_', ds) AS rand_ds
  SELECT  SPLIT_PART(rand_ds, '_', 2) AS ds
          ,COUNT(DISTINCT shop_id) AS id_cnt
  FROM (
      SELECT  CONCAT(CAST(FLOOR(RAND() * 10) AS STRING), '_', ds) AS rand_ds
              ,shop_id
      FROM    demo_data0
  )
  GROUP BY rand_ds;
  
  -- 方法 B：乱数フィールドを追加します。ROUND(RAND(),1)*10 AS randint10
  SELECT  ds
          ,COUNT(DISTINCT shop_id) AS id_cnt
  FROM    (SELECT  ds
                   ,shop_id
             FROM  demo_data0
          )
  GROUP BY ds, FLOOR(RAND() * 10);

• 方法 3：2 段階集計のような集計。

  GROUP BY フィールドと DISTINCT フィールドのデータが均等に分散している場合は、まず 2 つのグループ化フィールド (ds と shop_id) に GROUP BY を適用し、次に count(distinct) コマンドを使用することでクエリを最適化できます。

  SELECT  ds
          ,COUNT(shop_id) AS cnt
  FROM(SELECT  ds
              ,shop_id
              FROM    demo_data0
              GROUP BY ds ,shop_id
      )
  GROUP BY ds;

次のコードは、Top-10 の例です。

SELECT  main_id
        ,type
FROM    (SELECT  main_id
                 ,type
                 ,ROW_NUMBER() OVER(PARTITION BY main_id ORDER BY type DESC ) rn
            FROM <data_demo2>
        ) A
WHERE   A.rn <= 10;

データスキューが発生した場合、次のいずれかの方法で解決できます。

• 方法 1：SQL ベースの 2 段階集計。

  map ステージで各パーティショングループのデータをできるだけ均等に分散させるために、乱数列を追加し、それを PARTITION BY 句のパラメーターとして使用します。

  -- 方法 1：乱数にモジュロを使用します。
  SELECT  main_id
          ,type
    FROM  (SELECT  main_id
                   ,type
                   ,ROW_NUMBER() OVER(PARTITION BY main_id ORDER BY type DESC ) rn
              FROM (SELECT  main_id
                            ,type
                          FROM (SELECT  main_id
                                        ,type
                                        ,ROW_NUMBER() OVER(PARTITION BY main_id,src_pt ORDER BY type DESC ) rn
                                   FROM (SELECT  main_id
                                                 ,type
                                                 ,ceil(110 * rand()) % 11 AS src_pt
                                           FROM  data_demo2
                                        )
                                  ) B
                          WHERE   B.rn <= 10
                      )
          ) A
  WHERE   A.rn <= 10;
  -- 方法 2：カスタムの乱数を使用します。
  SELECT  main_id
          ,type
    FROM  (SELECT  main_id
                   ,type
                   ,ROW_NUMBER() OVER(PARTITION BY main_id ORDER BY type DESC ) rn
              FROM (SELECT  main_id
                            ,type
                      FROM(SELECT  main_id
                                   ,type
                                   ,ROW_NUMBER() OVER(PARTITION BY main_id,src_pt ORDER BY type DESC ) rn
                             FROM  (SELECT  main_id
                                            ,type
                                            ,ceil(10 * rand()) AS src_pt
                                            FROM    data_demo2
                                    )
                                  ) B
                          WHERE B.rn <= 10
                      )
          ) A
  WHERE   A.rn <= 10;

• 方法 2：UDAF ベースの 2 段階集計。

  SQL メソッドは、冗長で保守が困難なコードになる可能性があります。代わりに、UDAF と最小ヒープ優先度付きキューを使用して最適化できます。iterate フェーズでは Top-N 要素のみが保持され、merge フェーズでは N 要素のみがマージされます。プロセスは次のとおりです。

  • iterate：最初の K 個の要素をプッシュします。K 個より後の要素については、最小ヒープの先頭要素と継続的に比較し、必要に応じて要素を交換します。

  • merge：2 つのヒープをマージした後、上位 K 個の要素をその場で返します。

  • terminate：ヒープを配列として返します。

  • SQL クエリでは、配列を個別の行に分割します。

  @annotate('* -> array<string>')
  class GetTopN(BaseUDAF):
      def new_buffer(self):
          return [[], None]
      def iterate(self, buffer, order_column_val, k):
          # heapq.heappush(buffer, order_column_val)
          # buffer = [heapq.nlargest(k, buffer), k]
          if not buffer[1]:
              buffer[1] = k
          if len(buffer[0]) < k:
              heapq.heappush(buffer[0], order_column_val)
          else:
              heapq.heappushpop(buffer[0], order_column_val)
      def merge(self, buffer, pbuffer):
          first_buffer, first_k = buffer
          second_buffer, second_k = pbuffer
          k = first_k or second_k
          merged_heap = first_buffer + second_buffer
          merged_heap.sort(reverse=True)
          merged_heap = merged_heap[0: k] if len(merged_heap) > k else merged_heap
          buffer[0] = merged_heap
          buffer[1] = k
      def terminate(self, buffer):
          return buffer[0]
  
  SET odps.sql.python.version=cp37;
  SELECT main_id,type_val FROM (
    SELECT  main_id ,get_topn(type, 10) AS type_array
    FROM data_demo2
    GROUP BY main_id
  )
  LATERAL VIEW EXPLODE(type_array)type_ar AS type_val;

動的パーティションは多くのシナリオで使用され、データスキューを容易に引き起こす可能性があります。データスキューが発生した場合、次のいずれかの解決策で解決できます。

• 方法 1：パラメーター設定。

  動的パーティションは、異なる条件を満たすデータを異なるパーティションに配置できるため、複数の INSERT OVERWRITE ステートメントが不要になります。これにより、特にパーティションが多い場合にコードを大幅に簡素化できます。ただし、動的パーティションは、過剰な数の小さいファイルにつながる可能性もあります。

  • データスキューの例

    次の単純な SQL を例にとります。

    INSERT INTO TABLE part_test PARTITION(ds) SELECT * FROM  part_test;

    K 個の Map インスタンスと N 個のターゲットパーティションがあると仮定します。

                                ds=1
    cfile1                      ds=2
    ...             X           ds=3
    cfilek                      ...
                                ds=n

    最も極端なケースでは、K*N 個の小さいファイルが生成される可能性があります。過剰な数の小さいファイルは、ファイルシステムに大きな管理上の負荷をかける可能性があります。そのため、MaxCompute は、追加レベルの reducer タスクを導入することで動的パーティションを処理します。同じターゲットパーティションのデータを同じ (または少数の) reducer インスタンスによって書き込むように指示し、これにより、過剰な数の小さいファイルが作成されるのを回避します。この reducer は常にジョブの最後のタスクです。MaxCompute では、この機能はデフォルトで有効になっており、次のパラメーターが true に設定されていることを意味します。

    SET odps.sql.reshuffle.dynamicpt=true;

    この機能をデフォルトで有効にすると、小さいファイルが多すぎるという問題が解決され、単一のインスタンスによって生成されるファイルが多すぎるためにタスクが失敗するのを防ぎます。ただし、これにより、データスキューという新たな問題も発生します。さらに、追加の reducer ステージを導入すると、コンピューティングリソースが消費されます。したがって、トレードオフを慎重に検討する必要があります。

  • 解決策

    set odps.sql.reshuffle.dynamicpt=true; パラメーターを有効にして追加の reducer ステージを導入する当初の目的は、小さいファイルが多すぎるという問題を解決することでした。ただし、ターゲットパーティションの数が少なく、小さいファイルが多すぎるリスクがない場合、この機能をデフォルトで有効にすると、コンピューティングリソースを無駄にするだけでなく、パフォーマンスも低下します。この場合、set odps.sql.reshuffle.dynamicpt=false; を設定してこの機能を無効にすると、パフォーマンスが大幅に向上する可能性があります。次のコードに例を示します。

    INSERT overwrite TABLE ads_tb_cornucopia_pool_d PARTITION (ds, lv, tp)
    SELECT /*+ mapjoin(t2) */
        '20150503' AS ds,
        t1.lv AS lv,
        t1.type AS tp
    FROM
        (SELECT  ...
        FROM tbbi.ads_tb_cornucopia_user_d
        WHERE ds = '20150503'
        AND lv IN ('flat', '3rd')
        AND tp = 'T'
        AND pref_cat2_id > 0
        ) t1
    JOIN
        (SELECT ...
        FROM tbbi.ads_tb_cornucopia_auct_d
        WHERE ds = '20150503'
        AND tp = 'T'
        AND is_all = 'N'
        AND cat2_id > 0
        ) t2
    ON t1.pref_cat2_id = t2.cat2_id;

    上記のコードにデフォルトのパラメーターを使用した場合、ジョブの総ランタイムは約 1 時間 30 分です。最後の reducer ステージには約 1 時間 20 分かかり、これは総ランタイムの約 90% を占めます。追加の reducer ステージを導入すると、各 reducer インスタンスのデータ分布が非常に不均一になり、ロングテールが発生します。

  上記の例では、生成された動的パーティションの履歴数を分析すると、毎日約 2 つの動的パーティションしか生成されていないことがわかります。したがって、安全に set odps.sql.reshuffle.dynamicpt=false; を設定できます。これにより、ジョブはわずか 9 分で完了できます。この場合、このパラメーターを false に設定すると、パフォーマンスが大幅に向上し、コンピューティング時間とリソースを節約できます。この 1 つのパラメーター変更だけで、最小限の労力で大幅な改善が得られます。

  この最適化は、多くのリソースを消費する大規模で実行時間の長いジョブだけでなく、リソース消費の少ない通常の短時間実行ジョブにも適用されます。動的パーティションが使用され、動的パーティションの数が少ない限り、odps.sql.reshuffle.dynamicpt パラメーターを false に設定して、リソースを節約し、パフォーマンスを向上させることができます。

  ジョブの期間に関係なく、次の 3 つの条件をすべて満たすノードは最適化できます。

  • ジョブは動的パーティションを使用します。

  • 動的パーティションの数は 50 以下です。

  • ジョブに `set odps.sql.reshuffle.dynamicpt=false;` がありません。

  最後の Fuxi インスタンスの実行時間を使用して、ノードにこのパラメーターを設定する緊急度を判断できます。これは diag_level フィールドによって識別されます。ルールは次のとおりです。

  • Last_Fuxi_Inst_Time が 30 分を超える場合：Diag_Level=4 ('Critical')。

  • Last_Fuxi_Inst_Time が 20 分から 30 分の場合：Diag_Level=3 ('High')。

  • Last_Fuxi_Inst_Time が 10 分から 20 分の場合：Diag_Level=2 ('Medium')。

  • Last_Fuxi_Inst_Time が 10 分未満の場合：Diag_Level=1 ('Low')。

• 方法 2：プルーニング最適化。

  動的パーティションにデータを挿入する際に map ステージですでに存在するデータスキューを解決するには、レコード数が多いパーティションを見つけてプルーニングし、それらを個別に挿入します。実際のユースケースに基づいて、map ステージのパラメーター設定を次のように変更できます。

  SET odps.sql.mapper.split.size=128;
  INSERT OVERWRITE TABLE data_demo3 partition(ds,hh)
  SELECT  *
  FROM    dwd_alsc_ent_shop_info_hi;

  結果は、フルテーブルスキャンが実行されたことを示しています。さらに最適化するには、次のようにシステムによって導入された Reduce ジョブを無効にすることができます。

  SET odps.sql.reshuffle.dynamicpt=false ;
  INSERT OVERWRITE TABLE data_demo3 partition(ds,hh)
  SELECT *
  FROM dwd_alsc_ent_shop_info_hi;

  動的パーティションにデータを挿入する際に map ステージでデータスキューを解決するには、レコード数が多いパーティションを見つけてプルーニングし、それらを個別に挿入します。具体的な手順は次のとおりです。

  1. 次のコマンドを使用して、レコード数が多い特定のパーティションをクエリします。

     SELECT  ds
             ,hh
             ,COUNT(*) AS cnt
     FROM    dwd_alsc_ent_shop_info_hi
     GROUP BY ds
              ,hh
     ORDER BY cnt DESC;

     パーティションの一部を次に示します。

     ds	hh	cnt
     20200928	17	1052800
     20191017	17	1041234
     20210928	17	1034332
     20190328	17	1000321
     20210504	1	19
     20191003	20	18
     20200522	1	18
     20220504	1	18

  2. レコード数が多いパーティションをフィルターで除外し、残りのデータを挿入してから、レコード数が多いパーティションのデータを個別に挿入します。

     SET odps.sql.reshuffle.dynamicpt=false ;
     -- レコード数が多くないパーティションのデータを挿入します。
     INSERT OVERWRITE TABLE data_demo3 partition(ds,hh)
     SELECT  *
     FROM    dwd_alsc_ent_shop_info_hi
     WHERE   CONCAT(ds,hh) NOT IN ('2020092817','2019101717','2021092817','2019032817');
     
     -- レコード数が多いパーティションのデータを挿入します。
     set odps.sql.reshuffle.dynamicpt=false ;
     INSERT OVERWRITE TABLE data_demo3 partition(ds,hh)
     SELECT  *
     FROM    dwd_alsc_ent_shop_info_hi
     WHERE   CONCAT(ds,hh) IN ('2020092817','2019101717','2021092817','2019032817');
     
     -- 結果を検証します。
     SELECT  ds
       ,hh,COUNT(*) AS cnt
      FROM dwd_alsc_ent_shop_info_hi
      GROUP BY ds,hh ORDER BY cnt desc;