DQL operations SELECT DISTRIBUTED MAPJOIN - MaxCompute - Alibaba Cloud Documentation Center

DISTRIBUTED MAPJOIN merupakan versi teroptimalkan dari MAPJOIN yang dapat digunakan saat melakukan join antara tabel kecil dan tabel besar untuk mengurangi proses pengacakan (shuffling) dan pengurutan (sorting) pada tabel besar.

Catatan penggunaan

Ukuran tabel yang akan di-join harus berbeda: tabel besar harus melebihi 10 TB, sedangkan tabel kecil harus berada dalam rentang [1 GB, 100 GB].
Data dalam tabel kecil harus terdistribusi secara merata. Jika tabel kecil memiliki long tails, data berlebih akan terkonsentrasi pada satu shard, sehingga berpotensi menyebabkan error out of memory (OOM) atau timeout pada remote procedure call (RPC).
Jika task SQL berjalan lebih dari 20 menit, disarankan menggunakan DISTRIBUTED MAPJOIN untuk optimasi.
Task ini menggunakan sumber daya berlebih saat dijalankan, sehingga tidak disarankan untuk dijalankan dalam kelompok kuota kecil.

Catatan
Anda dapat mengubah kelompok kuota di halaman Quotas. Untuk informasi selengkapnya, lihat Manage quotas in the new MaxCompute console.

Gunakan DISTRIBUTED MAPJOIN

Untuk menggunakan DISTRIBUTED MAPJOIN dalam pernyataan SELECT, tambahkan petunjuk /*+distmapjoin(<table_name>(shard_count=<n>,replica_count=<m>))*/. Parameter shard_count dan replica_count digunakan untuk menentukan tingkat paralelisme task, yang dihitung dengan rumus: Parallelism = shard_count × replica_count.

Parameter
- table_name: nama tabel kecil yang akan di-join.
- shard_count=<n>: jumlah shard data dari tabel kecil yang didistribusikan ke setiap node komputasi untuk pemrosesan. Nilai n umumnya diatur ke angka ganjil.
  Catatan
  - Kami menyarankan agar Anda menentukan secara manual parameter shard_count. Nilai parameter shard_count dapat diperkirakan berdasarkan ukuran tabel kecil, dengan jumlah data yang diproses oleh satu shard node idealnya berada dalam rentang [200 MB, 500 MB].
  - Jika Anda mengatur parameter shard_count ke nilai yang terlalu besar, kinerja dan stabilitas pemrosesan data akan terganggu. Sebaliknya, jika nilai parameter shard_count terlalu kecil, error dapat terjadi akibat penggunaan memori yang berlebihan.
- replica_count=<m>: jumlah replika dari tabel kecil. Nilai m menentukan jumlah replika, dengan nilai default 1.
  Catatan
  Untuk mengurangi permintaan akses berlebih dan mencegah kegagalan seluruh task akibat kegagalan satu node, Anda dapat membuat beberapa replika dalam shard yang sama. Jika suatu node sering restart karena paralelisme tinggi atau ketidakstabilan lingkungan, pertimbangkan meningkatkan nilai replica_count. Disarankan mengatur parameter ini ke nilai 2 atau 3.

Sintaks

-- Direkomendasikan: Tentukan shard_count. replica_count default-nya 1.
/*+distmapjoin(a(shard_count=5))*/

-- Direkomendasikan: Tentukan shard_count dan replica_count.
/*+distmapjoin(a(shard_count=5,replica_count=2))*/

-- Gunakan DISTRIBUTED MAPJOIN untuk multiple tabel kecil.
/*+distmapjoin(a(shard_count=5,replica_count=2),b(shard_count=5,replica_count=2)) */

-- Gunakan DISTRIBUTED MAPJOIN dan MAPJOIN bersamaan.
/*+distmapjoin(a(shard_count=5,replica_count=2)),mapjoin(b)*/

Contoh

Contoh berikut menjelaskan cara menggunakan DISTRIBUTED MAPJOIN saat memasukkan data ke dalam tabel partisi bernama tmall_dump_lasttable.

Sintaks JOIN

insert OVERWRITE table tmall_dump_lasttable partition(ds='20211130')
select t1.*
from
(
    select nid, doc,type
    from search_ods.dump_lasttable where ds='20211203'
)t1
join
(
    select distinct item_id
    from tbcdm.dim_tb_itm
    where ds='20211130'
    and bc_type='B'
    and is_online='Y'
)t2
on t1.nid=t2.item_id;

Sintaks DISTRIBUTED MAPJOIN

insert OVERWRITE table tmall_dump_lasttable partition (ds='20211130')
select /*+ distmapjoin(t2(shard_count=35)) */ t1.*
from
(
    select nid, doc, type
    from search_ods.dump_lasttable where ds='20211203'
)t1
join
(
    select distinct item_id
    from tbcdm.dim_tb_itm
    where ds='20211130'
    and bc_type='B'
    and is_online='Y'
)t2
on t1.nid=t2.item_id;