DocsICP FilingConsole
Documentation
Home Page MaxCompute Developer Reference Engines MapReduce Instructions Run MapReduce jobs in SQL runtime

Run MapReduce jobs in SQL runtime

Updated at:
Copy as MD
Product
My Favorites

MaxCompute supports running MapReduce jobs in the sql runtime execution mode. Using the sql runtime, your MapReduce jobs can leverage new features from the SQL engine and enable previously unsupported functionality. This topic describes how to specify the sql runtime execution mode for a MapReduce job.

Background

MaxCompute provides a MapReduce API that you can use to write Java programs for processing data in MaxCompute.

The latest version of MaxCompute allows you to run MapReduce jobs in the sql runtime execution mode. In this mode, MapReduce jobs can use the MaxCompute SQL compiler, cost-based optimizer (CBO), and vectorized execution engine, benefiting from the continuous improvements made to the SQL engine in features, performance, and stability.

MapReduce jobs running in the sql runtime execution mode can use new MaxCompute SQL features to perform previously unsupported operations. Compared to the classic MapReduce engine, the sql runtime execution mode adds the following features:

  • Support for views as an input source.

  • Support for external tables as an input source.

  • Support for read and write operations on a distributed file system.

  • Support for read and write operations on hash- or range-clustered tables.

This mode also provides the following benefits:

  • Leverage the continuous performance optimizations of the SQL cost-based optimizer (CBO) and vectorized execution engine.

  • Use new storage format compression mechanisms.

  • Dynamically adjust parallelism to improve performance in scenarios that join very large tables, such as hash-clustered tables.

  • Leverage the SQL engine's proven stability, established through extensive job execution and stress testing. This improves the reliability of mechanisms like failover and persistent volume claim (PVC).

  • Gain deeper insights by using MaxCompute Studio and LogView. These tools provide detailed execution information for SQL jobs, such as the execution plan, compilation information, and job configurations. This detailed view of the I/O at each stage and the overall workflow helps you identify issues, optimize performance, and improve development and O&M efficiency.

Notes

  • This feature does not require any changes to your existing APIs or job logic. You only need to specify the execution mode.

  • This feature supports only MapReduce jobs written with the MapReduce API. For more information, see SDK Overview.

  • MapReduce jobs that run in the sql runtime execution mode are still billed according to the standard MapReduce billing rules. For more information, see MapReduce pay-as-you-go.

Instructions

  1. Set the execution mode.

    You can control the execution mode with the odps.mr.run.mode property. The following values are valid:

    • lot (Default): The job runs on the MapReduce execution engine.

    • sql: The job runs on the SQL execution engine. If the job fails, an error is returned.

    • hybrid: The system first tries to run the job on the SQL execution engine. If the attempt fails, the job falls back to the MapReduce execution engine.

    You can set the execution mode in one of the following ways:

    Project-level control

    To enable this mode for all jobs in a project, a project administrator must run the following command:

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

    Session-level control

    To enable this mode for the current job only, use one of the following methods:

    • Add the set odps.mr.run.mode=<lot/sql/hybrid> statement before the JAR command.

    • Set the property in your job code using a JobConf object, as shown in the following example:

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

    For special scenarios such as StreamJob and SecondarySort, you must set the following flags:

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

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

  2. View job details.

    You can use tools such as LogView and MaxCompute Studio to inspect the SQL expressions generated on the client and view the job's execution details. For more information about how to use LogView, see Use LogView V2.0 to view job information.

    • LogView XML

      Open LogView and go to the Source XML tab to view the XML information submitted by the client. This information shows the SQL equivalent of the MapReduce job. The following code provides an example:

      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 ;

    • LogView Summary

      On the Summary tab in LogView, you can see that the job ran using the SQL execution engine. The following code provides an example:

      Note

      For MapReduce jobs that do not run in the sql runtime execution mode, no execution engine information is displayed. For MaxCompute extended MapReduce (MR2) jobs that do not run in the sql runtime execution mode, the execution engine is cganjiang.

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

    • LogView JSON Summary

      The JSON Summary for a standard MapReduce job contains only basic input and output information for the Map and Reduce tasks. In contrast, the JSON Summary for a SQL-based job provides detailed information for each execution phase, including all execution parameters, the logical plan, the physical plan, and execution details. The following code provides an example:

       "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}]])
]
Previous：Local runNext：Examples