MaxCompute:Penyetelan kesenjangan data

Kode berikut memberikan contoh pseudo-code dengan klausa 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;

Saat terjadi kesenjangan data, Anda dapat menggunakan salah satu dari tiga solusi berikut:

• Metode 1: Atur parameter anti-skew untuk GROUP BY.

  SET odps.sql.groupby.skewindata=true;

• Metode 2: Tambahkan bilangan acak.

  Solusi ini menulis ulang SQL dengan menambahkan bilangan acak, memisahkan kunci yang menyebabkan long tail. Ini adalah metode efektif untuk mengatasi long tail dalam operasi GROUP BY.

  Untuk kueri SQL Select Key,Count(*) As Cnt From TableName Group By Key;, tanpa combiner, node mapper melakukan shuffle data ke node reducer, yang kemudian melakukan operasi COUNT. Rencana eksekusi yang sesuai adalah M->R.

  Dengan asumsi kunci long-tail telah diidentifikasi, Anda dapat mendistribusikan ulang pekerjaan untuk kunci tersebut sebagai berikut:

  -- Asumsikan kunci long-tail adalah 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;

  Rencana eksekusi yang dimodifikasi menjadi M->R->R. Meskipun jumlah langkah eksekusi bertambah, waktu proses keseluruhan mungkin berkurang karena kunci long-tail diproses dalam dua tahap. Konsumsi resource dan efisiensi waktu mirip dengan Metode 1. Namun, dalam skenario dunia nyata, sering kali ada lebih dari satu kunci long-tail. Mengingat upaya untuk menemukan kunci long-tail dan menulis ulang SQL, Metode 1 sering kali lebih hemat biaya.

• Metode 3: Buat tabel rolling.

  Untuk mengurangi biaya dan meningkatkan efisiensi, Anda mungkin perlu mengambil data dari tahun lalu. Untuk task online, membaca semua partisi dari T-1 hingga T-365 setiap kali merupakan pemborosan resource yang signifikan. Membuat tabel rolling dapat mengurangi jumlah partisi yang dibaca tanpa memengaruhi pengambilan data untuk tahun lalu. Kode berikut memberikan contohnya.

  Pertama, inisialisasi data bisnis merchant selama 365 hari dengan agregasi GROUP BY, tandai tanggal pembaruan data, dan simpan sebagai tabel a. Task online berikutnya kemudian dapat menggabungkan tabel T-2 a dengan tabel table_xxx_di dan melakukan GROUP BY lagi. Hal ini mengurangi jumlah partisi yang dibaca setiap hari dari 365 menjadi 2. Duplikasi primary key shop_id sangat berkurang, yang juga menurunkan konsumsi resource.

  -- Buat tabel rolling.
  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 'Partisi tanggal'
  )LIFECYCLE 7;
  -- Asumsikan periode 365 hari adalah 2021-05-01 hingga 2022-05-01. Lakukan inisialisasi satu kali.
  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;
  -- Kemudian, task online harian yang dijalankan adalah:
  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` -- Cegah rolling tak terbatas hari buka.
    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}'
        -- Tidak perlu GROUP BY di sini jika sumber sudah dikelompokkan.
      )
      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;
                              

Asumsikan sebuah tabel memiliki distribusi data sebagai berikut.

Menggunakan pernyataan berikut dapat dengan mudah menyebabkan kesenjangan data:

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

Solusinya adalah sebagai berikut:

• Metode 1: Penyetelan parameter.

  Atur parameter berikut:

  SET odps.sql.groupby.skewindata=true;

• Metode 2: Agregasi dua tahap generik.

  Jika data di field shop_id didistribusikan secara tidak merata, Metode 1 tidak efektif. Metode yang lebih generik adalah menambahkan bilangan acak ke nilai field partisi.

  -- Metode A: Gabungkan bilangan acak. 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;
  
  -- Metode B: Tambahkan field bilangan acak. 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);

• Metode 3: Agregasi mirip dua tahap.

  Jika data untuk field GROUP BY dan DISTINCT didistribusikan secara merata, Anda dapat mengoptimalkan kueri dengan terlebih dahulu menerapkan GROUP BY pada dua field pengelompokan (ds dan shop_id), lalu menggunakan perintah 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;

Kode berikut memberikan contoh 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;

Saat terjadi kesenjangan data, Anda dapat mengatasinya dengan salah satu metode berikut:

• Metode 1: Agregasi dua tahap berbasis SQL.

  Untuk mendistribusikan data di setiap grup partisi se-merata mungkin selama tahap map, tambahkan kolom acak dan gunakan sebagai parameter dalam klausa PARTITION BY.

  -- Metode 1: Gunakan modulo pada bilangan acak.
  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;
  -- Metode 2: Gunakan bilangan acak kustom.
  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;

• Metode 2: Agregasi dua tahap berbasis UDAF.

  Metode SQL dapat menghasilkan kode yang panjang dan sulit dipelihara. Sebagai alternatif, Anda dapat menggunakan UDAF dengan antrian prioritas min-heap untuk optimasi. Pada fase iterate, hanya elemen Top-N yang disimpan, dan pada fase merge, hanya N elemen yang digabungkan. Prosesnya sebagai berikut:

  • iterate: Dorong K elemen pertama. Untuk elemen setelah K, terus-menerus bandingkan dengan elemen teratas min-heap dan tukar elemen jika diperlukan.

  • merge: Setelah menggabungkan dua heap, kembalikan K elemen teratas di tempatnya.

  • terminate: Kembalikan heap sebagai array.

  • Dalam kueri SQL, pisahkan array menjadi baris-baris terpisah.

  @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 partition digunakan dalam banyak skenario dan dapat dengan mudah menyebabkan kesenjangan data. Saat terjadi kesenjangan data, Anda dapat mengatasinya dengan salah satu solusi berikut.

• Metode 1: Konfigurasi parameter.

  Partisi dinamis dapat menempatkan data yang memenuhi kondisi berbeda ke partisi berbeda, yang menghindari kebutuhan beberapa pernyataan INSERT OVERWRITE. Hal ini sangat menyederhanakan kode, terutama saat ada banyak partisi. Namun, partisi dinamis juga dapat menyebabkan jumlah file kecil yang berlebihan.

  • Contoh kesenjangan data

    Ambil SQL sederhana berikut sebagai contoh:

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

    Asumsikan ada K instance Map dan N partisi target.

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

    Dalam kasus paling ekstrem, K*N file kecil dapat dihasilkan. Jumlah file kecil yang berlebihan dapat memberikan tekanan manajemen besar pada sistem file. Oleh karena itu, MaxCompute menangani partisi dinamis dengan memperkenalkan level tambahan task reducer. Data untuk partisi target yang sama diarahkan untuk ditulis oleh instance reducer yang sama (atau beberapa), yang menghindari pembuatan terlalu banyak file kecil. Reducer ini selalu menjadi task terakhir dalam job. Di MaxCompute, fitur ini diaktifkan secara default, artinya parameter berikut diatur ke true:

    SET odps.sql.reshuffle.dynamicpt=true;

    Mengaktifkan fitur ini secara default menyelesaikan masalah terlalu banyak file kecil dan mencegah task gagal karena jumlah file yang dihasilkan oleh satu instance berlebihan. Namun, hal ini juga memperkenalkan masalah baru: kesenjangan data. Selain itu, memperkenalkan tahap reducer tambahan mengonsumsi resource komputasi. Oleh karena itu, Anda harus mempertimbangkan trade-off-nya dengan hati-hati.

  • Solusi

    Tujuan awal memperkenalkan tahap reducer tambahan dengan mengaktifkan parameter set odps.sql.reshuffle.dynamicpt=true; adalah untuk menyelesaikan masalah terlalu banyak file kecil. Namun, jika jumlah partisi target kecil dan tidak ada risiko memiliki terlalu banyak file kecil, mengaktifkan fitur ini secara default tidak hanya memboroskan resource komputasi tetapi juga mengurangi kinerja. Dalam kasus ini, menonaktifkan fitur ini dengan mengatur set odps.sql.reshuffle.dynamicpt=false; dapat secara signifikan meningkatkan kinerja. Kode berikut memberikan contohnya.

    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;

    Jika parameter default digunakan untuk kode di atas, waktu proses total job sekitar 1 jam 30 menit. Tahap reducer terakhir memakan waktu sekitar 1 jam 20 menit, yang menyumbang sekitar 90% dari total waktu proses. Pengenalan tahap reducer tambahan membuat distribusi data setiap instance reducer sangat tidak merata, yang menyebabkan long tail.

  Untuk contoh di atas, dengan menganalisis jumlah historis partisi dinamis yang dihasilkan, kami menemukan bahwa hanya sekitar dua partisi dinamis yang dihasilkan setiap hari. Oleh karena itu, Anda dapat dengan aman mengatur set odps.sql.reshuffle.dynamicpt=false;. Job kemudian dapat diselesaikan hanya dalam 9 menit. Dalam kasus ini, mengatur parameter ini ke false dapat secara signifikan meningkatkan kinerja serta menghemat waktu dan resource komputasi. Perubahan parameter tunggal ini memberikan peningkatan signifikan dengan usaha minimal.

  Optimasi ini tidak hanya untuk job besar yang berjalan lama dan mengonsumsi banyak resource, tetapi juga untuk job biasa yang berjalan singkat dan mengonsumsi sedikit resource. Selama menggunakan partisi dinamis dan jumlah partisi dinamis kecil, Anda dapat mengatur parameter odps.sql.reshuffle.dynamicpt ke false untuk menghemat resource dan meningkatkan kinerja.

  Node yang memenuhi ketiga kondisi berikut dapat dioptimalkan, terlepas dari durasi job:

  • Job menggunakan partisi dinamis.

  • Jumlah partisi dinamis 50 atau kurang.

  • Job tidak memiliki set odps.sql.reshuffle.dynamicpt=false;.

  Waktu proses instance Fuxi terakhir dapat digunakan untuk menentukan urgensi pengaturan parameter ini untuk node tersebut. Hal ini diidentifikasi oleh field diag_level. Aturannya sebagai berikut:

  • Last_Fuxi_Inst_Time lebih dari 30 menit: Diag_Level=4 ('Critical').

  • Last_Fuxi_Inst_Time antara 20 dan 30 menit: Diag_Level=3 ('High').

  • Last_Fuxi_Inst_Time antara 10 dan 20 menit: Diag_Level=2 ('Medium').

  • Last_Fuxi_Inst_Time kurang dari 10 menit: Diag_Level=1 ('Low').

• Metode 2: Optimasi pruning.

  Untuk mengatasi kesenjangan data yang sudah ada di tahap map saat memasukkan data ke partisi dinamis, Anda dapat menemukan dan melakukan pruning pada partisi dengan banyak catatan, lalu memasukkannya secara terpisah. Berdasarkan kasus penggunaan aktual, Anda dapat memodifikasi konfigurasi parameter tahap map sebagai berikut:

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

  Hasilnya menunjukkan bahwa pemindaian tabel penuh dilakukan. Untuk optimasi lebih lanjut, Anda dapat menonaktifkan job Reduce yang diperkenalkan oleh sistem, sebagai berikut:

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

  Untuk mengatasi kesenjangan data di tahap map saat memasukkan data ke partisi dinamis, temukan partisi dengan banyak catatan, lakukan pruning, lalu masukkan secara terpisah. Langkah-langkah spesifiknya adalah sebagai berikut:

  1. Gunakan perintah berikut untuk mencari partisi spesifik dengan jumlah catatan besar.

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

     Beberapa partisi tersebut adalah sebagai berikut:

     ds	hh	cnt
     20200928	17	1.052.800
     20191017	17	1.041.234
     20210928	17	1.034.332
     20190328	17	1.000.321
     20210504	1	19
     20191003	20	18
     20200522	1	18
     20220504	1	18

  2. Filter partisi dengan jumlah catatan besar, masukkan data yang tersisa, lalu masukkan secara terpisah data untuk partisi dengan catatan banyak.

     SET odps.sql.reshuffle.dynamicpt=false ;
     -- Masukkan data untuk partisi yang tidak memiliki jumlah catatan besar.
     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');
     
     -- Masukkan data untuk partisi yang memiliki jumlah catatan besar.
     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');
     
     -- Verifikasi hasilnya.
     SELECT  ds
       ,hh,COUNT(*) AS cnt
      FROM dwd_alsc_ent_shop_info_hi
      GROUP BY ds,hh ORDER BY cnt desc;