MaxCompute:Jalankan pekerjaan MapReduce dalam runtime SQL

Latar Belakang

MaxCompute menyediakan API MapReduce yang dapat Anda gunakan untuk menulis program Java guna memproses data di MaxCompute.

Versi terbaru MaxCompute memungkinkan Anda menjalankan pekerjaan MapReduce dalam mode eksekusi runtime SQL. Dalam mode ini, pekerjaan MapReduce dapat memanfaatkan kompilator SQL MaxCompute, optimasi berbasis biaya (CBO), serta mesin eksekusi vektorisasi, sehingga mendapatkan manfaat dari peningkatan berkelanjutan pada mesin SQL dalam hal fitur, kinerja, dan stabilitas.

Pekerjaan MapReduce yang dijalankan dalam mode eksekusi runtime SQL dapat menggunakan fitur-fitur SQL MaxCompute terbaru untuk melakukan operasi yang sebelumnya tidak didukung. Dibandingkan dengan mesin MapReduce klasik, mode eksekusi runtime SQL menambahkan fitur-fitur berikut:

• Dukungan untuk views sebagai sumber input.

• Dukungan untuk tabel eksternal sebagai sumber input.

• Dukungan untuk operasi baca dan tulis pada sistem file terdistribusi.

• Dukungan untuk operasi baca dan tulis pada tabel terkluster hash atau range.

Mode ini juga memberikan manfaat-manfaat berikut:

• Memanfaatkan optimasi kinerja berkelanjutan dari CBO SQL dan mesin eksekusi vektorisasi.

• Menggunakan mekanisme kompresi format penyimpanan baru.

• Menyesuaikan paralelisme secara dinamis untuk meningkatkan kinerja dalam skenario yang melibatkan join tabel sangat besar, seperti tabel terkluster hash.

• Mendapatkan stabilitas terbukti dari mesin SQL yang telah diuji melalui eksekusi pekerjaan dan uji stres secara ekstensif. Hal ini meningkatkan keandalan mekanisme seperti failover dan klaim volume persisten (PVC).

• Memperoleh wawasan lebih mendalam dengan menggunakan MaxCompute Studio dan LogView. Alat-alat ini menyediakan informasi eksekusi terperinci untuk pekerjaan SQL, seperti rencana eksekusi, informasi kompilasi, dan konfigurasi pekerjaan. Tampilan terperinci mengenai I/O pada setiap tahap dan alur kerja secara keseluruhan membantu Anda mengidentifikasi masalah, mengoptimalkan kinerja, serta meningkatkan efisiensi pengembangan dan O&M.

Catatan

• Fitur ini tidak memerlukan perubahan apa pun pada API atau logika pekerjaan yang sudah ada. Anda hanya perlu menentukan mode eksekusi.

• Fitur ini hanya mendukung pekerjaan MapReduce yang ditulis dengan API MapReduce. Untuk informasi selengkapnya, lihat Ikhtisar SDK.

• Pekerjaan MapReduce yang dijalankan dalam mode eksekusi runtime SQL tetap ditagih sesuai aturan penagihan MapReduce standar. Untuk informasi selengkapnya, lihat MapReduce pay-as-you-go.

Petunjuk

1. Tentukan mode eksekusi.

   Anda dapat mengontrol mode eksekusi melalui properti odps.mr.run.mode. Nilai-nilai berikut valid:

   • lot (Default): Pekerjaan dijalankan pada mesin eksekusi MapReduce.

   • sql: Pekerjaan dijalankan pada mesin eksekusi SQL. Jika pekerjaan gagal, error akan dikembalikan.

   • hybrid: Sistem pertama-tama mencoba menjalankan pekerjaan pada mesin eksekusi SQL. Jika upaya tersebut gagal, pekerjaan akan dialihkan ke mesin eksekusi MapReduce.

   Anda dapat menentukan mode eksekusi dengan salah satu cara berikut:

   Kontrol tingkat Proyek

   Untuk mengaktifkan mode ini bagi semua pekerjaan dalam suatu Proyek, administrator Proyek harus menjalankan perintah berikut:

   setproject odps.mr.run.mode=<lot/sql/hybrid>;

   Kontrol tingkat Sesi

   Untuk mengaktifkan mode ini hanya untuk pekerjaan saat ini, gunakan salah satu metode berikut:

   • Tambahkan pernyataan set odps.mr.run.mode=<lot/sql/hybrid> sebelum perintah JAR.

   • Tentukan properti dalam kode pekerjaan Anda menggunakan objek JobConf, seperti pada contoh berikut:

     JobConf job = new JobConf();
     job.set("odps.mr.run.mode","hybrid")

   Catatan

   Untuk skenario khusus seperti StreamJob dan SecondarySort, Anda harus menyetel flag-flag berikut:

   • StreamJob: set odps.mr.sql.stream.enable=true;

   • SecondarySort: set odps.mr.sql.group.enable=true;

2. Lihat detail pekerjaan.

   Anda dapat menggunakan alat seperti LogView dan MaxCompute Studio untuk memeriksa ekspresi SQL yang dihasilkan di sisi klien dan melihat detail eksekusi pekerjaan tersebut. Untuk informasi selengkapnya tentang cara menggunakan LogView, lihat Gunakan LogView V2.0 untuk melihat informasi pekerjaan.

   • LogView XML

     Buka LogView dan buka tab Source XML untuk melihat informasi XML yang dikirimkan oleh klien. Informasi ini menunjukkan ekuivalen SQL dari pekerjaan MapReduce. Kode berikut memberikan contohnya:

     create temporary function mr2sql_mapper_152955927079392291755 as   'com.aliyun.odps.mapred.bridge.LotMapperUDTF' using ; 
     create temporary function mr2sql_reducer_152955927079392291755 as 'com.aliyun.odps.mapred.bridge.LotReducerUDTF' using ; 
     
     @sub_query_mapper :=
     SELECT k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code
     FROM(
       SELECT mr2sql_mapper_152955927079392291755(id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code ) as (k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code)
       FROM ae_antispam.product_sku_tt_inc
       WHERE ds = "20180615"  AND hh = "21"                     
       UNION ALL
       SELECT mr2sql_mapper_152955927079392291755(id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code ) as (k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code)
       FROM ae_antispam.product_sku
     ) open_mr_alias1
     DISTRIBUTE BY k_id SORT BY k_id ASC;
     
     @sub_query_reducer := 
     SELECT mr2sql_reducer_152955927079392291755(k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code) as (id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code)
     FROM @sub_query_mapper;
     FROM @sub_query_reducer	
     INSERT OVERWRITE TABLE ae_antispam.product_sku
     SELECT id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code ;

   • Ringkasan LogView

     Pada tab Summary di LogView, Anda dapat melihat bahwa pekerjaan tersebut dijalankan menggunakan execution engine SQL. Kode berikut memberikan contohnya:

     Catatan

     Untuk pekerjaan MapReduce yang tidak dijalankan dalam mode eksekusi sql runtime, informasi execution engine tidak ditampilkan. Untuk pekerjaan extended MapReduce (MR2) MaxCompute yang tidak dijalankan dalam mode eksekusi sql runtime, execution engine-nya adalah cganjiang.

      Job run mode: fuxi job
      Job run engine: execution engine

   • Ringkasan JSON LogView

     JSON Summary untuk pekerjaan MapReduce standar hanya berisi informasi input dan output dasar untuk tugas Map dan Reduce. Sebaliknya, JSON Summary untuk pekerjaan berbasis SQL menyediakan informasi rinci untuk setiap fase eksekusi, termasuk semua parameter eksekusi, logical plan, physical plan, dan detail eksekusi. Kode berikut memberikan contohnya:

      "midlots" : 
      [
      "LogicalTableSink(table=[[odps_flighting.flt_20180621104445_step1_ad_quality_tech_qp_algo_antifake_wordbag_filter_bag_change_result_lv2_20, auctionid,word,match_word(3) {0, 1, 2}]])
     OdpsLogicalProject(auctionid=[$0], word=[$1], match_word=[$2])
     OdpsLogicalProject(auctionid=[$0], word=[$1], match_word=[$2])
     OdpsLogicalProject(auctionid=[$0], word=[$1], match_word=[$2])
     OdpsLogicalProject(auctionid=[$2], word=[$3], match_word=[$4])
     OdpsLogicalTableFunctionScan(invocation=[[MR2SQL_MAPPER_152955294118813063732($0, $1)]()], rowType=[RecordType(VARCHAR(2147483647) item_id, VARCHAR(2147483647) text, VARCHAR(2147483647) __tf_0_0, VARCHAR(2147483647) __tf_0_1, VARCHAR(2147483647) __tf_0_2)])
     OdpsLogicalTableScan(table=[[ad_quality_tech.qp_algo_antifake_wordbag_filter_bag_change_lv2_20, item_id,text(2) {0, 1}]])
     ]