MaxCompute:Data skew tuning

The following code provides a pseudo-code example with a GROUP BY clause.

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

When data skew occurs, you can use one of the following three solutions:

• Method 1: Set the anti-skew parameter for GROUP BY.

  SET odps.sql.groupby.skewindata=true;

• Method 2: Add a random number.

  This solution rewrites the SQL to add a random number, splitting the keys that cause long tails. This is an effective method for resolving long tails in GROUP BY operations.

  For the SQL query Select Key,Count(*) As Cnt From TableName Group By Key;, without a combiner, the mapper node shuffles data to the reducer node, which then performs the COUNT operation. The corresponding execution plan is M->R.

  Assuming the long-tail key has been identified, you can re-distribute the work for that key as follows:

  -- Assume the long-tail key is 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;

  The modified execution plan becomes M->R->R. Although the number of execution steps increases, the overall runtime may be reduced because the long-tail key is processed in two stages. The resource consumption and time efficiency are similar to Method 1. However, in real-world scenarios, there is often more than one long-tail key. Considering the effort of finding long-tail keys and rewriting SQL, Method 1 is often more cost-effective.

• Method 3: Create a rolling table.

  To reduce costs and improve efficiency, you may need to retrieve data from the past year. For online tasks, reading all partitions from T-1 to T-365 each time is a significant waste of resources. Creating a rolling table can reduce the number of partitions read without affecting data retrieval for the past year. The following code provides an example.

  First, initialize 365 days of merchant business data with a GROUP BY aggregation, mark the data update date, and store it as table a. Subsequent online tasks can then join the T-2 day table a with the table_xxx_di table and perform another GROUP BY. This reduces the number of partitions read daily from 365 to 2. The duplication of the primary key shop_id is greatly reduced, which also decreases resource consumption.

  -- Create a rolling table.
  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;
  -- Assume the 365-day period is 2021-05-01 to 2022-05-01. Perform a one-time initialization.
  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;
  -- Then, the daily online task to be executed is:
  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` -- Prevent infinite rolling of 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}'
        -- No GROUP BY needed here if the source is already grouped.
      )
      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;
                              

Suppose a table has the following data distribution.

Using the following statement can easily cause data skew:

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

The solutions are as follows:

• Method 1: Parameter tuning.

  Set the following parameter:

  SET odps.sql.groupby.skewindata=true;

• Method 2: Generic two-stage aggregation.

  If the data in the shop_id field is unevenly distributed, Method 1 is not effective. A more generic method is to append a random number to the partition field value.

  -- Method A: Concatenate a random number. 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;
  
  -- Method B: Add a random number field. 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);

• Method 3: Two-stage-like aggregation.

  If the data for the GROUP BY and DISTINCT fields is evenly distributed, you can optimize the query by first applying GROUP BY to the two grouping fields (ds and shop_id) and then using the count(distinct) command.

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

The following code provides a Top-10 example.

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;

When data skew occurs, you can resolve it by using one of the following methods:

• Method 1: SQL-based two-stage aggregation.

  To distribute the data in each partition group as evenly as possible during the map stage, add a random column and use it as a parameter in the PARTITION BY clause.

  -- Method 1: Use modulo on a random number.
  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;
  -- Method 2: Use a custom random number.
  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;

• Method 2: UDAF-based two-stage aggregation.

  The SQL method can result in verbose code that is difficult to maintain. Alternatively, you can use a UDAF with a min-heap priority queue for optimization. In the iterate phase, only the Top-N elements are kept, and in the merge phase, only N elements are merged. The process is as follows:

  • iterate: Push the first K elements. For elements after K, continuously compare them with the top element of the min-heap and swap elements as needed.

  • merge: After merging two heaps, return the top K elements in place.

  • terminate: Return the heap as an array.

  • In the SQL query, split the array into separate rows.

  @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;

Dynamic partitions are used in many scenarios and can easily lead to data skew. When data skew occurs, you can resolve it by using one of the following solutions.

• Method 1: Parameter configuration.

  Dynamic partitioning can place data that meets different conditions into different partitions, which avoids the need for multiple INSERT OVERWRITE statements. This can greatly simplify the code, especially when there are many partitions. However, dynamic partitioning can also lead to an excessive number of small files.

  • Data skew example

    Take the following simple SQL as an example:

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

    Assume there are K Map instances and N target partitions.

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

    In the most extreme case, K*N small files may be generated. An excessive number of small files can put huge management pressure on the file system. Therefore, MaxCompute handles dynamic partitions by introducing an additional level of reducer tasks. It directs data for the same target partitions to be written by the same (or a few) reducer instances, which avoids creating too many small files. This reducer is always the last task in the job. In MaxCompute, this feature is enabled by default, which means the following parameter is set to true:

    SET odps.sql.reshuffle.dynamicpt=true;

    Enabling this feature by default solves the problem of too many small files and prevents tasks from failing due to an excessive number of files generated by a single instance. However, it also introduces a new problem: data skew. Additionally, introducing an extra reducer stage consumes computing resources. Therefore, you must carefully weigh the trade-offs.

  • Solution

    The initial purpose of introducing an extra reducer stage by enabling the set odps.sql.reshuffle.dynamicpt=true; parameter is to solve the problem of too many small files. However, if the number of target partitions is small and there is no risk of having too many small files, enabling this feature by default not only wastes computing resources but also reduces performance. In this case, disabling this feature by setting set odps.sql.reshuffle.dynamicpt=false; can significantly improve performance. The following code provides an example.

    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;

    If the default parameters are used for the preceding code, the total runtime of the job is about 1 hour and 30 minutes. The last reducer stage takes about 1 hour and 20 minutes, which accounts for about 90% of the total runtime. The introduction of an extra reducer stage makes the data distribution of each reducer instance very uneven, which leads to a long tail.

  For the preceding example, by analyzing the historical number of dynamic partitions generated, we find that only about two dynamic partitions are generated each day. Therefore, you can safely set set odps.sql.reshuffle.dynamicpt=false;. The job can then be completed in just 9 minutes. In this case, setting this parameter to false can significantly improve performance and save computing time and resources. This single parameter change provides a significant improvement for minimal effort.

  This optimization is not only for large, long-running jobs that consume a lot of resources, but also for ordinary, short-running jobs that consume fewer resources. As long as dynamic partitioning is used and the number of dynamic partitions is small, you can set the odps.sql.reshuffle.dynamicpt parameter to false to save resources and improve performance.

  Nodes that meet all the following three conditions can be optimized, regardless of the job's duration:

  • The job uses dynamic partitions.

  • The number of dynamic partitions is 50 or less.

  • The job does not have set odps.sql.reshuffle.dynamicpt=false;.

  The execution time of the last Fuxi instance can be used to determine the urgency of setting this parameter for the node. This is identified by the diag_level field. The rules are as follows:

  • Last_Fuxi_Inst_Time is greater than 30 minutes: Diag_Level=4 ('Critical').

  • Last_Fuxi_Inst_Time is between 20 and 30 minutes: Diag_Level=3 ('High').

  • Last_Fuxi_Inst_Time is between 10 and 20 minutes: Diag_Level=2 ('Medium').

  • Last_Fuxi_Inst_Time is less than 10 minutes: Diag_Level=1 ('Low').

• Method 2: Pruning optimization.

  To resolve data skew that already exists in the map stage when inserting data into dynamic partitions, you can find and prune the partitions with many records, then insert them separately. Based on the actual use case, you can modify the parameter configuration of the map stage as follows:

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

  The result shows that a full table scan was performed. To further optimize, you can disable the Reduce job introduced by the system, as follows:

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

  To resolve data skew in the map stage when inserting data into dynamic partitions, find the partitions with many records, prune them, and insert them separately. The specific steps are as follows:

  1. Use the following command to query for specific partitions with a large number of records.

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

     Some of the partitions are as follows:

     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. Filter out the partitions with a large number of records, insert the remaining data, and then separately insert the data for the large-record partitions.

     SET odps.sql.reshuffle.dynamicpt=false ;
     -- Insert data for partitions that do not have a large number of records.
     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');
     
     -- Insert data for partitions that have a large number of records.
     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');
     
     -- Verify the result.
     SELECT  ds
       ,hh,COUNT(*) AS cnt
      FROM dwd_alsc_ent_shop_info_hi
      GROUP BY ds,hh ORDER BY cnt desc;