MaxCompute:Gabungkan file kecil

Cara kerja

MaxCompute menggunakan dua mekanisme untuk mengendalikan akumulasi file kecil:

Automatic merging — Setelah job selesai, MaxCompute mengalokasikan task Fuxi untuk menggabungkan file kecil ketika jumlah file dalam suatu partisi melebihi ambang batas. Secara default, satu instans Fuxi dapat memproses maksimal 100 file kecil. Anda dapat mengenali proses merge otomatis melalui entri MergeTask dalam log job. MaxCompute juga secara berkala melakukan pemindaian terhadap metadatabase dan menggabungkan file kecil di tabel atau partisi mana pun yang memiliki jumlah file tinggi.

Manual merging — Ketika data ditulis terus-menerus ke suatu tabel dan penggabungan otomatis tidak mampu mengimbanginya, hentikan job penulisan tersebut dan jalankan perintah MERGE SMALLFILES secara manual.

Batasan

• Menggabungkan file kecil mengonsumsi sumber daya komputasi. Pada instans pay-as-you-go, biaya dikenakan dengan tarif yang sama seperti penagihan SQL pay-as-you-go. Untuk detail harga, lihat Computing pricing (pay-as-you-go).

• Perintah MERGE SMALLFILES tidak mendukung tabel transaksional. Untuk melakukan compact pada tabel transaksional, gunakan perintah COMPACTION sebagai gantinya.

Periksa jumlah file dalam tabel

Jalankan DESC EXTENDED untuk melihat berapa banyak file yang dimiliki oleh suatu tabel atau partisi.

Sintaksis

DESC EXTENDED <table_name> [PARTITION (<pt_spec>)];

Parameter

Contoh output

Contoh di atas menunjukkan tabel odl_bpm_wfc_task_log: 3.607 file, ukuran total 274 MB (287.869.658 byte), rata-rata ukuran file ~0,07 MB. Partisi ini merupakan kandidat yang baik untuk digabungkan.

Kapan harus menggabungkan: Jika suatu partisi berisi lebih dari 100 file dan ukuran file rata-ratanya di bawah 64 MB, gabungkan file-file kecil tersebut.

Gabungkan file kecil

Menggunakan perintah SQL

Jalankan perintah berikut untuk menggabungkan file kecil sesuai permintaan:

ALTER TABLE <table_name> [PARTITION (<pt_spec>)] MERGE SMALLFILES;

Setelah penggabungan, jumlah file pada tabel odl_bpm_wfc_task_log turun dari 3.607 menjadi 19 file, dan penyimpanan menyusut dari 274 MB menjadi 37 MB.

Parameter default sudah cukup untuk sebagian besar kasus. Gunakan parameter SET berikut untuk menyesuaikan perilaku penggabungan jika diperlukan:

Parameter	Deskripsi	Default
odps.merge.cross.paths	Apakah akan menggabungkan file lintas path. Atur ke true untuk menggabungkan file kecil di setiap path secara terpisah tanpa mengubah struktur path.	—
odps.merge.smallfile.filesize.threshold	File dengan ukuran (dalam MB) sama dengan atau di bawah nilai ini memenuhi syarat untuk digabungkan. Jika tidak diatur, menggunakan variabel global odps_g_merge_filesize_threshold (default: 32 MB). Atur ke nilai lebih besar dari 32 saat mengganti nilai default.	32 MB
odps.merge.maxmerged.filesize.threshold	Ukuran maksimum (dalam MB) dari satu file output setelah penggabungan. File output baru dibuat ketika batas ini tercapai. Jika tidak diatur, menggunakan variabel global odps_g_max_merged_filesize_threshold (default: 500 MB). Atur ke nilai lebih besar dari 500 saat mengganti nilai default.	500 MB
odps.merge.max.filenumber.per.instance	Jumlah maksimum file yang dapat diproses oleh satu instans merge.	—
odps.sql.mapper.merge.limit.size	Ukuran total maksimum file yang dibaca oleh satu instans Fuxi.	—

import os
from odps import ODPS

# Muat kredensial dari variabel lingkungan.
# Jangan hardcode ID AccessKey atau Rahasia AccessKey Anda dalam kode.
o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='<your-endpoint>',
)

# Ganti <table_name> dengan nama tabel Anda.
table_name = '<table_name>'
t = o.get_table(table_name)

# Atur opsi merge.
hints = {'odps.merge.maxmerged.filesize.threshold': 256}

# Kirim satu task merge per partisi di bawah tanggal target.
# Ganti <datetime> dengan nilai tanggal yang dituju, misalnya '20240101'.
insts = []
for partition in t.iterate_partitions(spec='ds=<datetime>'):
    instance = o.run_merge_files(table_name, str(partition), hints=hints)
    # Buka URL Logview ini dan klik Waiting Queue untuk menemukan log job merge.
    print(instance.get_logview_address())
    insts.append(instance)

# Tunggu hingga semua task merge selesai.
for inst in insts:
    inst.wait_for_completion()

Contoh

Pemeriksaan stabilitas data mengidentifikasi bahwa tbcdm.dwd_tb_log_pv_di perlu digabungkan. Kueri terhadap tabel metadata tbcdm.dws_rmd_merge_task_1d menunjukkan bahwa sebagian besar partisi melebihi 1.000 file, beberapa bahkan melebihi 7.000 file, dan beberapa partisi memiliki ukuran file rata-rata di bawah 1 MB.

Jalankan perintah berikut untuk menggabungkan file kecil:

SET odps.merge.cross.paths=true;
SET odps.merge.smallfile.filesize.threshold=128;
SET odps.merge.max.filenumber.per.instance = 2000;
ALTER TABLE tbcdm.dwd_tb_log_pv_di PARTITION (ds='20151116') MERGE SMALLFILES;

Hasil setelah penggabungan:

Langkah selanjutnya

• COMPACTION — Compact data dalam tabel transaksional

• Computing pricing (pay-as-you-go) — Pahami penagihan job merge