MaxCompute:Optimisation de la répartition inégale des données

Le code suivant fournit un exemple de pseudo-code avec une clause 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;

Lorsqu'une répartition inégale des données se produit, vous pouvez utiliser l'une des trois solutions suivantes :

• Méthode 1 : Définition du paramètre anti-déséquilibre pour GROUP BY.

  SET odps.sql.groupby.skewindata=true;

• Méthode 2 : Ajout d'un nombre aléatoire.

  Cette solution réécrit le SQL pour ajouter un nombre aléatoire, séparant ainsi les clés responsables des longues traînes. Il s'agit d'une méthode efficace pour résoudre les longues traînes dans les opérations GROUP BY.

  Pour la requête SQL Select Key,Count(*) As Cnt From TableName Group By Key;, sans combiner, le nœud mapper redistribue les données vers le nœud reducer, qui effectue ensuite l'opération COUNT. Le plan d'exécution correspondant est M->R.

  En supposant que la clé à longue traîne ait été identifiée, vous pouvez redistribuer le travail pour cette clé comme suit :

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

  Le plan d'exécution modifié devient M->R->R. Bien que le nombre d'étapes d'exécution augmente, le temps d'exécution global peut être réduit car la clé à longue traîne est traitée en deux étapes. La consommation de ressources et l'efficacité temporelle sont similaires à celles de la Méthode 1. Cependant, dans des scénarios réels, il existe souvent plus d'une clé à longue traîne. Compte tenu de l'effort nécessaire pour trouver les clés à longue traîne et réécrire le SQL, la Méthode 1 est souvent plus rentable.

• Méthode 3 : Création d'une table tournante.

  Pour réduire les coûts et améliorer l'efficacité, vous devrez peut-être récupérer des données sur l'année écoulée. Pour les tâches en ligne, lire toutes les partitions de T-1 à T-365 à chaque fois représente un gaspillage important de ressources. La création d'une table tournante peut réduire le nombre de partitions lues sans affecter la récupération des données sur l'année écoulée. Le code suivant fournit un exemple.

  Tout d'abord, initialisez 365 jours de données commerciales marchandes avec une agrégation GROUP BY, marquez la date de mise à jour des données et stockez-les dans la table a. Les tâches en ligne ultérieures peuvent alors joindre la table du jour T-2 a avec la table table_xxx_di et effectuer un autre GROUP BY. Cela réduit le nombre de partitions lues quotidiennement de 365 à 2. La duplication de la clé primaire shop_id est grandement réduite, ce qui diminue également la consommation de ressources.

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

Supposons qu'une table présente la distribution de données suivante.

L'utilisation de l'instruction suivante peut facilement provoquer une répartition inégale des données :

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

Les solutions sont les suivantes :

• Méthode 1 : Réglage des paramètres.

  Définissez le paramètre suivant :

  SET odps.sql.groupby.skewindata=true;

• Méthode 2 : Agrégation générique en deux étapes.

  Si les données du champ shop_id sont réparties de manière inégale, la Méthode 1 n'est pas efficace. Une méthode plus générique consiste à ajouter un nombre aléatoire à la valeur du champ de partition.

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

• Méthode 3 : Agrégation de type deux étapes.

  Si les données des champs GROUP BY et DISTINCT sont réparties uniformément, vous pouvez optimiser la requête en appliquant d'abord GROUP BY aux deux champs de regroupement (ds et shop_id), puis en utilisant la commande 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;

Le code suivant fournit un exemple 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;

Lorsqu'une répartition inégale des données se produit, vous pouvez la résoudre en utilisant l'une des méthodes suivantes :

• Méthode 1 : Agrégation en deux étapes basée sur SQL.

  Pour distribuer les données de chaque groupe de partition aussi uniformément que possible lors de l'étape map, ajoutez une colonne aléatoire et utilisez-la comme paramètre dans la clause PARTITION BY.

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

• Méthode 2 : Agrégation en deux étapes basée sur UDAF.

  La méthode SQL peut entraîner un code verbeux difficile à maintenir. Vous pouvez alternativement utiliser une UDAF avec une file de priorité min-heap pour l'optimisation. Durant la phase iterate, seuls les éléments Top-N sont conservés, et durant la phase merge, seuls N éléments sont fusionnés. Le processus est le suivant :

  • iterate : Poussez les K premiers éléments. Pour les éléments suivants, comparez-les continuellement à l'élément supérieur du min-heap et échangez les éléments si nécessaire.

  • merge : Après la fusion de deux tas, renvoyez les K premiers éléments sur place.

  • terminate : Renvoyez le tas sous forme de tableau.

  • Dans la requête SQL, divisez le tableau en lignes distinctes.

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

CREATE TABLE total_revenues (revenue bigint) partitioned BY (region string);

INSERT overwrite TABLE total_revenues PARTITION(region)
SELECT total_price AS revenue,region
FROM sale_detail;

Les partitions dynamiques sont utilisées dans de nombreux scénarios et peuvent facilement entraîner une répartition inégale des données. Lorsqu'une répartition inégale des données se produit, vous pouvez la résoudre en utilisant l'une des solutions suivantes.

• Méthode 1 : Configuration des paramètres.

  Le partitionnement dynamique permet de placer les données répondant à différentes conditions dans des partitions distinctes, évitant ainsi la nécessité de multiples instructions INSERT OVERWRITE. Cela peut grandement simplifier le code, surtout lorsqu'il y a de nombreuses partitions. Cependant, le partitionnement dynamique peut également entraîner un nombre excessif de petits fichiers.

  • Exemple de répartition inégale des données

    Prenons l'exemple du SQL simple suivant :

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

    Supposons qu'il existe K instances Map et N partitions cibles.

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

    Dans le cas le plus extrême, K*N petits fichiers peuvent être générés. Un nombre excessif de petits fichiers peut exercer une énorme pression de gestion sur le système de fichiers. Par conséquent, MaxCompute gère les partitions dynamiques en introduisant un niveau supplémentaire de tâches reducer. Il dirige les données destinées aux mêmes partitions cibles pour qu'elles soient écrites par la même instance reducer (ou quelques-unes), évitant ainsi la création de trop nombreux petits fichiers. Ce reducer est toujours la dernière tâche du job. Dans MaxCompute, cette fonctionnalité est activée par défaut, ce qui signifie que le paramètre suivant est défini sur true :

    SET odps.sql.reshuffle.dynamicpt=true;

    L'activation de cette fonctionnalité par défaut résout le problème du nombre excessif de petits fichiers et empêche les échecs de tâches dus à un nombre trop important de fichiers générés par une seule instance. Cependant, elle introduit également un nouveau problème : la répartition inégale des données. De plus, l'introduction d'une étape reducer supplémentaire consomme des ressources de calcul. Vous devez donc peser soigneusement les compromis.

  • Solution

    L'objectif initial de l'introduction d'une étape reducer supplémentaire en activant le paramètre set odps.sql.reshuffle.dynamicpt=true; était de résoudre le problème du nombre excessif de petits fichiers. Cependant, si le nombre de partitions cibles est faible et qu'il n'y a aucun risque d'avoir trop de petits fichiers, activer cette fonctionnalité par défaut non seulement gaspille des ressources de calcul, mais réduit également les performances. Dans ce cas, désactiver cette fonctionnalité en définissant set odps.sql.reshuffle.dynamicpt=false; peut améliorer considérablement les performances. Le code suivant fournit un exemple.

    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;

    Si les paramètres par défaut sont utilisés pour le code précédent, le temps d'exécution total du job est d'environ 1 heure et 30 minutes. La dernière étape reducer prend environ 1 heure et 20 minutes, soit environ 90% du temps d'exécution total. L'introduction d'une étape reducer supplémentaire rend la distribution des données de chaque instance reducer très inégale, ce qui entraîne une longue traîne.

  Pour l'exemple précédent, en analysant le nombre historique de partitions dynamiques générées, nous constatons qu'environ deux partitions dynamiques sont générées chaque jour. Vous pouvez donc définir en toute sécurité set odps.sql.reshuffle.dynamicpt=false;. Le job peut alors être terminé en seulement 9 minutes. Dans ce cas, définir ce paramètre sur false peut améliorer considérablement les performances et économiser du temps de calcul et des ressources. Ce changement de paramètre unique apporte une amélioration significative pour un effort minimal.

  Cette optimisation ne concerne pas uniquement les jobs volumineux et longs qui consomment beaucoup de ressources, mais aussi les jobs ordinaires et courts qui en consomment moins. Tant que le partitionnement dynamique est utilisé et que le nombre de partitions dynamiques est faible, vous pouvez définir le paramètre odps.sql.reshuffle.dynamicpt sur false pour économiser des ressources et améliorer les performances.

  Les nœuds remplissant les trois conditions suivantes peuvent être optimisés, quelle que soit la durée du job :

  • Le job utilise des partitions dynamiques.

  • Le nombre de partitions dynamiques est inférieur ou égal à 50.

  • Le job ne contient pas set odps.sql.reshuffle.dynamicpt=false;.

  Le temps d'exécution de la dernière instance Fuxi peut être utilisé pour déterminer l'urgence de définir ce paramètre pour le nœud. Ceci est identifié par le champ diag_level. Les règles sont les suivantes :

  • Last_Fuxi_Inst_Time est supérieur à 30 minutes : Diag_Level=4 ('Critical').

  • Last_Fuxi_Inst_Time est compris entre 20 et 30 minutes : Diag_Level=3 ('High').

  • Last_Fuxi_Inst_Time est compris entre 10 et 20 minutes : Diag_Level=2 ('Medium').

  • Last_Fuxi_Inst_Time est inférieur à 10 minutes : Diag_Level=1 ('Low').

• Méthode 2 : Optimisation par élagage.

  Pour résoudre la répartition inégale des données déjà présente dans l'étape map lors de l'insertion de données dans des partitions dynamiques, vous pouvez identifier et élaguer les partitions contenant de nombreux enregistrements, puis les insérer séparément. Selon le cas d'utilisation réel, vous pouvez modifier la configuration des paramètres de l'étape map comme suit :

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

  Le résultat montre qu'une analyse complète de la table a été effectuée. Pour optimiser davantage, vous pouvez désactiver le job Reduce introduit par le système, comme suit :

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

  Pour résoudre la répartition inégale des données dans l'étape map lors de l'insertion de données dans des partitions dynamiques, identifiez les partitions contenant de nombreux enregistrements, élaguez-les et insérez-les séparément. Les étapes spécifiques sont les suivantes :

  1. Utilisez la commande suivante pour interroger les partitions spécifiques contenant un grand nombre d'enregistrements.

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

     Certaines des partitions sont les suivantes :

     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. Filtrez les partitions contenant un grand nombre d'enregistrements, insérez les données restantes, puis insérez séparément les données des partitions à grand nombre d'enregistrements.

     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;