MaxCompute：Range clustering

Range clustering membagi data menjadi beberapa rentang yang saling terpisah berdasarkan pengurutan penuh kunci kluster. Setiap rentang dianggap sebagai bucket dan harus memenuhi kedua kondisi berikut:

• Nilai duplikat disimpan di bucket yang sama.

• Jumlah nilai di setiap bucket mendekati sama.

Pernyataan contoh berikut membuat sebuah tabel bernama T.

CREATE TABLE T (C1 int) 
    RANGE CLUSTERED BY (C1) 
    SORTED BY (c1) 
    INTO 3 BUCKETS;

Nilai di kolom C1 adalah { 1, 8, -3, 2, 4, 1, 1, 3, 8, 20, -8, 9 }.

Setelah Range Clustering diaktifkan, bucket berikut diperoleh:

• bucket 0 : { -8, -3, 1, 1, 1 }

• bucket 1 : { 2, 3, 4 }

• bucket 2 : { 8, 8, 9, 20 }

Proses range clustering dilaksanakan secara otomatis oleh MaxCompute. Anda tidak perlu menentukan setiap rentang secara manual. Dalam skenario big data, konfigurasi manual rentang tidak efisien atau memungkinkan. MaxCompute secara otomatis mengurutkan dan mengambil sampel data, membuat histogram berdasarkan distribusi data setiap rentang, kemudian menggabungkan dan menghitung histogram setiap rentang. Dengan cara ini, MaxCompute mencapai kinerja optimal dari range clustering.

Saat Anda membuat tabel, Anda dapat menentukan baik RANGE CLUSTERED BY maupun SORTED BY untuk memastikan bahwa data diurutkan secara global. Kemudian, MaxCompute secara otomatis membuat dua tingkat indeks: indeks global dan indeks file. Indeks digunakan untuk dengan cepat menemukan dan mencari nilai kunci, seperti yang ditunjukkan pada gambar berikut.

Range clustering memiliki keuntungan berikut dibandingkan pengelompokan hash:

• Kueri rentang didukung.

  Sebagai contoh, jika kondisi kueri adalah c < 3, sistem dapat mengecualikan bucket 2 dan bucket 3 berdasarkan indeks global, dan kemudian mengkueri data dari bucket 0 dan bucket 1. Untuk pengelompokan hash, bucket pruning hanya dapat dilakukan untuk kueri bersifat equality.

• Kueri multi-kunci didukung.

  Sebagai contoh, jika Anda menentukan RANGE CLUSTERED BY (c1, c2, c3) SORTED BY (c1, c2, c3) saat Anda membuat tabel, range clustering dan penyimpanan data dilakukan dalam urutan c1, c2, dan c3. Dengan cara ini, Anda dapat mengkueri data tabel berdasarkan kondisi yang kompleks, seperti c1 = 100 AND c2 > 0 atau c1 = 100 AND c2 = 50 AND c3 < 5. Jenis kueri ini tidak dapat dilakukan berdasarkan pengelompokan hash.

  Penting

  Untuk kueri berdasarkan beberapa kunci, kunci dalam kondisi kueri harus diurutkan sesuai urutan dan hanya kunci terakhir yang dapat digunakan untuk menentukan nilai rentang.

• Range clustering menyediakan implementasi pengurutan global yang efisien.

  Sebelum range clustering digunakan, MaxCompute hanya dapat menggunakan satu instance untuk mengurutkan data secara global, mengakibatkan efisiensi menjadi rendah. Setelah range clustering digunakan, data di setiap rentang dapat diurutkan secara bersamaan dan kemudian digabungkan, sehingga meningkatkan efisiensi secara signifikan.

Anda dapat menggunakan pernyataan CREATE TABLE untuk membuat tabel yang dikelompokkan berdasarkan rentang. Dalam pernyataan ini, Anda harus menentukan parameter RANGE CLUSTERED BY. Parameter INTO number_of_buckets BUCKETS dan SORTED BY bersifat opsional. Dalam sebagian besar kasus, kami sarankan Anda menentukan nilai yang sama di SORTED BY dan RANGE CLUSTERED BY. Dengan cara ini, efek optimalisasi optimal dapat dicapai.

• Sintaks

  CREATE TABLE [IF NOT EXISTS] <table_name>
               [(<col_name> data_type [comment <col_comment>], ...)]
               [comment table_comment]
               [PARTITIONED BY (<col_name> data_type [comment <col_comment>], ...)]
               [RANGE CLUSTERED BY (<col_name> [, <col_name>, ...])
               [SORTED BY (<col_name> [ASC | DESC]
               [, <col_name> [ASC | DESC] ...])]
               [INTO <number_of_buckets> BUCKETS]]
               [AS select_statement]

• Contoh

  • Tabel non-partisi

    CREATE TABLE T1 (a string, b string, c int)
                 RANGE CLUSTERED BY (c)
                 SORTED by (c)
                 INTO 1024 BUCKETS;

  • Tabel partisi

    CREATE TABLE T1 (a string, b string, c int)
                 PARTITIONED BY (dt int)
                 RANGE CLUSTERED BY (c)
                 SORTED by (c)
                 INTO 1024 BUCKETS;

• Parameter

  • RANGE CLUSTERED BY

    Menentukan kunci untuk range clustering. Setelah Anda menentukan parameter ini, MaxCompute akan mengurutkan dan melakukan sampling pada satu atau beberapa kolom data ke dalam rentang yang sesuai berdasarkan jumlah bucket yang Anda tentukan. Untuk mencegah masalah data skew dan hot spot serta meningkatkan kinerja kueri bersamaan, kami sarankan Anda untuk menentukan kolom dengan rentang nilai yang besar dan sedikit nilai kunci duplikat di RANGE CLUSTERED BY. Untuk mengoptimalkan kinerja kueri, kami sarankan Anda untuk menentukan kunci agregat atau kunci filter yang umum digunakan di RANGE CLUSTERED BY.

  • SORTED BY

    Menentukan cara mengurutkan field di bucket. Untuk mencapai kinerja kueri optimal, kami sarankan Anda menentukan nilai yang sama di SORTED BY dan RANGE CLUSTERED BY. Setelah Anda menentukan SORTED BY, MaxCompute secara otomatis akan menghasilkan indeks global dan indeks file serta menggunakan indeks tersebut untuk mempercepat kueri.

  • INTO number_of_buckets BUCKETS

    Berbeda dengan pengelompokan hash, INTO number_of_buckets BUCKETS bersifat opsional untuk range clustering. Jika Anda tidak menentukan parameter ini, MaxCompute akan secara otomatis menentukan jumlah bucket berdasarkan jumlah data. Dalam sebagian besar kasus, kami sarankan Anda menentukan jumlah bucket berdasarkan situasi aktual.

    Mirip dengan pengelompokan hash, untuk range clustering kami juga sarankan Anda menetapkan jumlah bucket berdasarkan ukuran bucket (512 MB hingga 1 GB). Untuk tabel yang sangat besar, diperlukan jumlah bucket yang banyak. Namun, kami sarankan agar jumlah bucket di tabel tidak melebihi 4.000.

Untuk tabel yang dipartisi, Anda dapat menjalankan pernyataan ALTER TABLE untuk menambah atau menghapus properti range clustering.

• Sintaks

  -- Ubah tabel menjadi tabel range-clustered.
  ALTER TABLE <table_name> [RANGE CLUSTERED BY (<col_name> [, <col_name>, ...])
                           [SORTED BY (<col_name> [ASC | DESC] [, <col_name> [ASC | DESC] ...])]
                           [INTO <number_of_buckets> BUCKETS];
  -- Ubah tabel range-clustered menjadi tabel non-range-clustered.
  ALTER TABLE <table_name> NOT CLUSTERED;

• Catatan Penggunaan

  • Pernyataan ALTER TABLE hanya dapat memodifikasi properti pengelompokan tabel partisi. Untuk tabel yang tidak dipartisi, properti pengelompokan tidak dapat diubah setelah properti tersebut ditambahkan ke tabel.

  • Pernyataan ALTER TABLE hanya berlaku untuk partisi baru tabel, yang mencakup partisi baru yang dihasilkan dengan menggunakan pernyataan INSERT OVERWRITE. Partisi baru disimpan berdasarkan properti pengelompokan. Format penyimpanan partisi yang sudah ada tetap tidak berubah.

  • Pernyataan ALTER TABLE hanya berlaku untuk partisi baru tabel. Oleh karena itu, Anda tidak dapat menentukan partisi dalam pernyataan ini.

Pernyataan ALTER TABLE cocok untuk tabel yang sudah ada. Setelah properti range clustering ditambahkan, partisi baru disimpan berdasarkan properti range clustering.

Setelah Anda membuat tabel range-clustered, Anda dapat menjalankan pernyataan berikut untuk melihat properti tabel. Properti range clustering ditampilkan di Extended Infodari hasil yang dikembalikan.

DESC EXTENDED <table_name>;

Gambar berikut menunjukkan contoh hasil yang dikembalikan. Untuk tabel yang dipartisi, Anda juga dapat menjalankan pernyataan berikut untuk melihat properti pengelompokan partisi.

DESC EXTENDED <table_name> partition(<pt_spec>);

Gambar berikut menunjukkan contoh hasil yang dikembalikan.

Sebagai contoh, untuk kondisi kueri sederhana id < 3, sistem mengekstrak kondisi dari pengoptimal dan mengonversi kondisi tersebut menjadi rentang nilai (-∞, 3). Dalam hal ini, sistem dapat menggunakan indeks global untuk bucket pruning guna mengecualikan bucket 2 dan bucket 3 yang datanya tidak berada dalam rentang nilai sebelumnya. Kemudian, sistem dapat menggunakan indeks setiap file di bucket 0 dan bucket 1 untuk secara cepat menemukan data yang dimaksud. Proses ini disebut penurunan predikat, seperti yang ditunjukkan pada gambar berikut.Berikut adalah contoh pernyataan TPC-H Query 6 yang digunakan untuk mengkueri data dari dataset 100 GB setelah range clustering dilakukan pada dataset tersebut. Dalam pernyataan TPC-H Query 6, operasi agregasi dilakukan berdasarkan penyaringan rentang. Range clustering dapat menggunakan dua tingkat indeks untuk secara cepat menemukan data. Dengan cara ini, durasi eksekusi kueri dan konsumsi sumber daya CPU dan memori dapat dikurangi secara signifikan.

select sum(l_extendedprice * l_discount) as revenue
  from tpch_lineitem l
 where l_shipdate >= '1994-01-01'
   and l_shipdate < '1995-01-01'
   and l_discount >= 0.05
   and l_discount <= 0.07
   and l_quantity < 24;

Dalam contoh ini, pernyataan berikut digunakan untuk mengubah tabel mf_tab menjadi tabel range-clustered. Hal ini membantu Anda memahami kueri multi-kunci dengan lebih baik.

ALTER TABLE mf_project.mf_tab 
            RANGE CLUSTERED BY (project_name, name)
            SORTED BY (project_name, name)
            INTO 1024 BUCKETS;

Setelah tabel diubah menjadi range-clustered, Anda dapat melakukan kueri agregat di tingkat proyek. Pernyataan sampel:

SELECT COUNT(*)
  from mf_project.mf_tab
 WHERE project_name="xxxdw"
   AND ds="20180115"
   AND type="TABLE";

Anda juga dapat menggunakan beberapa kunci untuk menentukan lokasi tabel secara tepat. Pernyataan sampel:

SELECT count(*)
  from mf_project.mf_tab
 WHERE project_name="xxxdw"
   AND name="adm_ctu_cle_kba_midun_trade_dd"
   AND type="TABLE";

Anda juga dapat menggunakan beberapa kunci untuk kueri rentang. Pernyataan berikut mengkueri tabel yang namanya diawali dengan adm.

SELECT count(*)
  from mf_project.mf_tab
 WHERE project_name="xxxdw"
   AND name>="adm"
   AND name < "adn"
   AND type="TABLE";

Semua kueri yang disebutkan dapat sepenuhnya memanfaatkan fitur pengurutan global range clustering dan melakukan penurunan predikat untuk mengurangi jumlah operasi I/O dalam pemindaian tabel serta menghemat sumber daya CPU dan memori yang dikonsumsi untuk penyaringan dan komputasi data.

Sebagai contoh, k1, k2 adalah kunci kluster di tabel bernama T.

• Jika kondisi kueri adalah k1 < 5, akselerasi kueri berbasis indeks dapat dicapai.

• Jika kondisi kueri adalah k1 = 10 AND k2 = 20, akselerasi kueri berbasis indeks dapat dicapai.

• Jika kondisi kueri adalah k1 = 10 AND k2< 0, akselerasi kueri berbasis indeks dapat dicapai.

• Jika kondisi kueri adalah k2 < 0, akselerasi kueri berbasis indeks tidak dapat dicapai. Ini karena k1 tidak ditentukan dalam kondisi kueri.

• Jika kondisi kueri adalah k1 < 0 AND k2 > 0, percepatan kueri berbasis indeks dapat digunakan untuk memperoleh data yang memenuhi kondisi k1 < 0. Untuk data yang memenuhi kondisi k2 > 0, pemindaian tabel diperlukan.

Sebagai contoh, sebuah tabel bernama T dibuat dengan menggunakan pernyataan CREATE TABLE berikut. Jika Anda mengkueri data dari tabel tersebut, Anda dapat melakukan operasi GROUP BY pada data tabel di tahap map.

CREATE TABLE T (department int, team string, employee string)
       RANGE CLUSTERED BY (department, team)
       SORTED BY (c1, c2)
       INTO 1024 BUCKETS;

SELECT COUNT(*) from T GROUP BY department, team;

Pernyataan berikut menunjukkan struktur data tabel foo.

create table foo(a bigint, b bigint, c bigint)
       range clustered by (a,b)
       sorted by(a,b) into 3 buckets;

Data yang disimpan di bucket tabel foo termasuk dalam rentang berikut:

Bucket 0: [1,1 : 3,3]
Bucket 1: [5,5 : 7,7]
Bucket 2: [8,8 : 9,9]

Rentang bucket yang disebutkan sebelumnya ditentukan dalam format Bucket N: [nilai batas bawah : nilai batas atas]. Jika data diagregasi berdasarkan Kolom a, rentang bucket berikut digunakan sebagai gantinya:

Bucket 0: [1 : 3]
Bucket 1: [5 : 7]
Bucket 2: [8 : 9]

Anda dapat langsung menghasilkan rencana eksekusi untuk operasi agregat berdasarkan Kolom a dan Kolom b, mulai tiga instance untuk mengagregatkan data di setiap bucket, dan kemudian mengembalikan hasil keluaran.

Namun, jika nilai Kolom a didistribusikan di beberapa bucket, hasil tidak valid dikembalikan. Contoh:

Bucket 0: [1,1 : 3,3]
Bucket 1: [3,5 : 7,7]
Bucket 2: [7,8 : 9,9]

Kolom a memiliki dua nilai, yaitu 3 dan 7, yang disimpan secara terpisah di dua bucket. Untuk mendapatkan hasil yang valid, Anda harus menempatkan tupel yang memiliki nilai yang sama di Kolom a ke dalam instance yang sama dan mengagregatkan tupel-tupel tersebut. Dengan cara ini, bucket akan dibuat lagi, seperti yang ditunjukkan pada gambar berikut. Spasi antara dua garis putus-putus merah menunjukkan rentang data yang dapat dibaca oleh setiap instance.Histogram diperlukan untuk range clustering. Untuk tabel range-clustered, jika kunci kluster dan kunci sort sama, pekerja yang berkorespondensi dengan setiap bucket mengambil sampel sebuah tuple untuk setiap 10.000 baris untuk mendapatkan nilai kunci kluster saat Anda memasukkan data ke tabel tersebut. Nilai yang diperoleh disimpan dalam histogram. Histogram setiap bucket disimpan di file metadata kluster. Jenis histogram ini disebut equi-depth histogram.

Setelah histogram setiap bucket diperoleh, Anda dapat membuat ulang bucket untuk setiap pekerja berdasarkan aturan berikut:

• Tupel yang memiliki kunci pengelompokan yang sama disimpan di bucket yang sama.

• Data didistribusikan secara merata di seluruh bucket.

Berdasarkan nilai batas bawah setiap bucket baru, setiap pekerja dapat membaca data dalam rentang yang valid dan mengembalikan hasil yang valid.

Konten berikut menggunakan tabel partsupp dengan data 1 TB dalam dataset TPC-H untuk menguji peningkatan kinerja. Jalankan pernyataan berikut untuk mengubah tabel partsupp menjadi tabel range-clustered:

CREATE TABLE partsupp ( PS_PARTKEY BIGINT NOT NULL,
                        PS_SUPPKEY BIGINT NOT NULL,
                        PS_AVAILQTY BIGINT NOT NULL,
                        PS_SUPPLYCOST  DECIMAL(15,2)  NOT NULL,
                        PS_COMMENT     VARCHAR(199) NOT NULL)
             RANGE CLUSTERED BY(PS_PARTKEY, PS_SUPPKEY)
             SORTED BY(PS_PARTKEY, PS_SUPPKEY) INTO 128 BUCKETS;

Jalankan pernyataan kueri berikut untuk melakukan pengujian:

SELECT ps_partkey, count(*) c FROM partsupp GROUP BY ps_partkey;

• Jalankan perintah berikut untuk menonaktifkan optimisasi:

   set odps.optimizer.enable.range.partial.repartitioning=false;

  Contoh output:

• Jalankan perintah berikut untuk mengaktifkan optimisasi:

  set odps.optimizer.enable.range.partial.repartitioning=true;

  Contoh output:

Hasil pengujian menunjukkan bahwa kecepatan kueri meningkat sebesar 57%, utilisasi CPU menurun sebesar 52%, dan penggunaan memori berkurang sebesar 71% setelah optimisasi diaktifkan. Peningkatan kinerja bervariasi tergantung pada jumlah data dan jenis kueri.

Secara default, akselerasi pengurutan global dinonaktifkan untuk tabel range-clustered. Untuk mengaktifkan akselerasi pengurutan global, Anda dapat menjalankan perintah berikut:

set odps.optimizer.distribute.ordering.enable=true;

Dibandingkan dengan hash clustering, range clustering memiliki batasan berikut:

• Biaya pembuatan data range clustering lebih tinggi daripada hash clustering. Hash clustering hanya merupakan operasi sederhana untuk hashing dan pengurutan data. Namun, untuk range clustering, pengambilan sampel data, pengurutan, dan penggabungan histogram diperlukan. Konsumsi keseluruhan, termasuk durasi eksekusi, biaya CPU, dan biaya memori, lebih tinggi dibandingan dengan konsumsi keseluruhan hash clustering. Oleh karena itu, jika hash clustering dapat digunakan untuk menyelesaikan masalah, Anda tidak perlu menggunakan range clustering.

• Range clustering tidak didukung dalam DYNAMIC PARTITION atau INSERT INTO.

• Range clustering hanya didukung untuk operasi join berikut: inner join, left outer join, right outer join, dan semi join. Range Clustering tidak didukung untuk anti-join atau full outer join.

• Untuk tabel range-clustered, kunci yang ditentukan dalam RANGE CLUSTERED BY harus sama dengan yang ditentukan dalam SORTED BY. Sebagai contoh, jika range clustered by (a,b) sorted by (a,b) ditentukan untuk tabel bernama foo saat Anda membuat tabel, optimalisasi yang dijelaskan dalam topik ini dapat dicapai pada tabel. Namun, jika range clustered by(a,b) sorted by (b,a) ditentukan untuk tabel bernama bar, optimalisasi yang dijelaskan dalam topik ini tidak dapat dicapai pada tabel.

• Kunci yang ditentukan dalam JOIN atau GROUP BY harus merupakan awalan atau semua kunci yang ditentukan dalam RANGE CLUSTERED BY. Sebagai contoh, range clustered by(a,b,c) sorted by(a,b,c) ditentukan dalam pernyataan pembuatan tabel. Optimalisasi yang dijelaskan dalam topik ini dapat dicapai pada tabel hanya jika a, a,b, atau a,b,c ditentukan sebagai kunci dalam JOIN atau GROUP BY. Optimalisasi yang dijelaskan dalam topik ini tidak dapat dicapai jika b atau a,c ditentukan sebagai kunci dalam JOIN atau GROUP BY.

• Untuk tabel range-clustered terpatisi, jika Anda ingin membaca data dari dua atau lebih partisi tabel, optimalisasi yang dijelaskan dalam topik ini tidak dapat dicapai. Optimalisasi yang dijelaskan dalam topik ini hanya dapat dicapai pada tabel terpatisi yang memiliki partisi tunggal dan tabel non-partisi.